-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy path_common.py
More file actions
47 lines (36 loc) · 1.79 KB
/
Copy path_common.py
File metadata and controls
47 lines (36 loc) · 1.79 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
"""Bits shared between framework integration modules.
Lives here so importing one framework integration doesn't drag in another's
dependencies.
"""
import os
import time
_CROCKFORD = '0123456789abcdefghjkmnpqrstvwxyz'
# Crockford base32, lowercase. 48-bit ms timestamp + 80-bit random = 128 bits → 26 chars.
def ulid():
n = (int(time.time() * 1000) << 80) | int.from_bytes(os.urandom(10), 'big')
out = []
for _ in range(26):
n, r = divmod(n, 32)
out.append(_CROCKFORD[r])
return ''.join(reversed(out))
# RAHM_LOG_TRACE_ID=disabled turns off trace_id binding + response-header echo in
# HTTP middlewares. App code that explicitly binds trace_id is unaffected.
# Parsed at import time of each framework module so per-module reload() picks
# up env-var changes in tests.
def parse_trace_id_setting():
setting = os.environ.get('RAHM_LOG_TRACE_ID', 'enabled').lower()
if setting not in ('enabled', 'disabled'):
raise ValueError('Incorrect value in the RAHM_LOG_TRACE_ID environment variable')
return setting == 'enabled'
# RAHM_LOG_QUIET_PATHS lists request paths whose successful HTTP access entry
# is logged at debug instead of info, so health-check / metrics polling doesn't
# flood the logs. Comma-separated; matched exactly. Parsed at import time of
# each framework module so reload() picks up env-var changes in tests.
def parse_quiet_paths():
raw = os.environ.get('RAHM_LOG_QUIET_PATHS', '/_health')
return frozenset(p.strip() for p in raw.split(',') if p.strip())
# Whether a completed request's access entry should drop from info to debug.
# Only non-error responses on a quiet path are downgraded — a failing health
# check (>=400) still surfaces at info.
def is_quiet_access(path, status, quiet_paths):
return status < 400 and path in quiet_paths