A Tracer Decorator With Arg Redaction (Python)
A `@trace()` decorator I bolt onto Python services when the production logs go quiet at the wrong layer. Logs entry, exit, duration, and exceptions, with secret-arg redaction baked in.
By @elisehuang
March 17, 2026
·
Updated May 20, 2026
862 views
6
4.4 (12)
from __future__ import annotations
import functools
import inspect
import time
from typing import Any, Callable
REDACT_PARAM_NAMES = {'password', 'token', 'api_key', 'secret', 'card_number'}
def trace(redact: set[str] | None = None) -> Callable[..., Any]:
redact_set = (redact or set()) | REDACT_PARAM_NAMES
def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
sig = inspect.signature(fn)
@functools.wraps(fn)
def wrapper(*args: Any, **kwargs: Any) -> Any:
bound = sig.bind_partial(*args, **kwargs)
bound.apply_defaults()
scrubbed = {
k: ('[REDACTED]' if k in redact_set else _safe_repr(v))
for k, v in bound.arguments.items()
}
t0 = time.perf_counter()
print(f'[trace] enter {fn.__qualname__}({scrubbed})')
try:
result = fn(*args, **kwargs)
dt = (time.perf_counter() - t0) * 1000
print(f'[trace] exit {fn.__qualname__} ok dt={dt:.1f}ms')
return result
except Exception as exc:
dt = (time.perf_counter() - t0) * 1000
print(f'[trace] exit {fn.__qualname__} raised {type(exc).__name__} dt={dt:.1f}ms')
raise
return wrapper
return decorator
def _safe_repr(v: Any) -> str:
s = repr(v)
return s if len(s) <= 80 else s[:80] + f'...({len(s)})'
@trace()
def login(user: str, password: str) -> dict:
return {'user': user, 'session': 'abc123'}
@trace(redact={'card_number'})
def charge(user: str, amount: int, card_number: str) -> str:
return f'charged {amount} to ****{card_number[-4:]}'
login('[email protected]', password='hunter2')
charge('[email protected]', 4200, card_number='4242424242424242')I keep this in a tracer.py next to my dotfile pretty-printer. The trick is inspect.signature(fn).bind_partial(*args, **kwargs), which gives me a name-keyed view of arguments regardless of how the caller passed them. That lets the redact set work on password whether the caller did login('me', 'hunter2') or login('me', password='hunter2'). The _safe_repr helper truncates long blobs so a single 50KB request body does not bury the rest of the log. Drop the decorator on a suspect function during an incident and remove it after the postmortem.
from __future__ import annotations
import functools
import inspect
import json
import time
import uuid
from contextvars import ContextVar
from typing import Any, Callable
_current_span = ContextVar('current_span', default=None)
_depth: ContextVar[int] = ContextVar('depth', default=0)
REDACT = {'password', 'token', 'api_key'}
def trace(redact: set[str] | None = None) -> Callable[..., Any]:
extra = (redact or set()) | REDACT
def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
sig = inspect.signature(fn)
@functools.wraps(fn)
def wrapper(*args: Any, **kwargs: Any) -> Any:
parent = _current_span.get()
span_id = uuid.uuid4().hex[:8]
depth = _depth.get()
t1 = _current_span.set(span_id)
t2 = _depth.set(depth + 1)
try:
bound = sig.bind_partial(*args, **kwargs).arguments
args_clean = {k: '[REDACTED]' if k in extra else repr(v) for k, v in bound.items()}
start = time.perf_counter()
event = {'evt': 'enter', 'fn': fn.__qualname__, 'span': span_id,
'parent': parent, 'depth': depth, 'args': args_clean}
print(json.dumps(event))
try:
out = fn(*args, **kwargs)
dur = (time.perf_counter() - start) * 1000
print(json.dumps({'evt': 'exit', 'fn': fn.__qualname__,
'span': span_id, 'ms': round(dur, 2), 'ok': True}))
return out
except Exception as exc:
dur = (time.perf_counter() - start) * 1000
print(json.dumps({'evt': 'exit', 'fn': fn.__qualname__,
'span': span_id, 'ms': round(dur, 2), 'ok': False,
'err': type(exc).__name__}))
raise
finally:
_current_span.reset(t1)
_depth.reset(t2)
return wrapper
return decorator
@trace()
def db_query(sql: str) -> list:
return [{'id': 1}]
@trace()
def get_user(user_id: int) -> dict:
rows = db_query(f'SELECT * FROM users WHERE id={user_id}')
return rows[0]
@trace(redact={'pin'})
def authenticate(email: str, pin: str) -> dict:
return get_user(42)
authenticate('[email protected]', pin='4242')Once I have more than one decorated function in the call chain, line-by-line logs lose context: which enter matches which exit? contextvars.ContextVar solves it cleanly because it follows asyncio task boundaries (unlike thread-locals). Each call mints a new span_id, captures the previous one as parent, and the structured-log shipper can reconstruct the call tree later without OpenTelemetry. I have used this in production at two startups while we were still six months from being able to justify a full tracing vendor.
from __future__ import annotations
import asyncio
import functools
import inspect
import time
from typing import Any, Callable
REDACT = {'password', 'token', 'api_key'}
def trace_async(redact: set[str] | None = None) -> Callable[..., Any]:
extra = (redact or set()) | REDACT
def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
sig = inspect.signature(fn)
is_coro = asyncio.iscoroutinefunction(fn)
async def async_wrap(*args, **kwargs):
bound = sig.bind_partial(*args, **kwargs).arguments
clean = {k: '[REDACTED]' if k in extra else repr(v) for k, v in bound.items()}
t0 = time.perf_counter()
print(f'[trace] enter async {fn.__qualname__}({clean})')
try:
result = await fn(*args, **kwargs)
print(f'[trace] exit async {fn.__qualname__} ok dt={(time.perf_counter()-t0)*1000:.1f}ms')
return result
except Exception as exc:
print(f'[trace] exit async {fn.__qualname__} raised {type(exc).__name__}')
raise
def sync_wrap(*args, **kwargs):
bound = sig.bind_partial(*args, **kwargs).arguments
clean = {k: '[REDACTED]' if k in extra else repr(v) for k, v in bound.items()}
t0 = time.perf_counter()
print(f'[trace] enter {fn.__qualname__}({clean})')
try:
result = fn(*args, **kwargs)
print(f'[trace] exit {fn.__qualname__} ok dt={(time.perf_counter()-t0)*1000:.1f}ms')
return result
except Exception as exc:
print(f'[trace] exit {fn.__qualname__} raised {type(exc).__name__}')
raise
return functools.wraps(fn)(async_wrap if is_coro else sync_wrap)
return decorator
@trace_async()
async def fetch_user(user_id: int, token: str) -> dict:
await asyncio.sleep(0.01)
return {'id': user_id}
@trace_async()
def parse_token(token: str) -> str:
return token[:4]
async def main():
await fetch_user(7, token='secret')
parse_token('abc.def.ghi')
asyncio.run(main())The synchronous version above wraps coroutines incorrectly: it logs exit immediately because fn(*args, **kwargs) returns a coroutine object, not the awaited result. The fix is to detect coroutine functions with asyncio.iscoroutinefunction(fn) and pick a sync or async wrapper accordingly. I always run the async branch through await so the duration reflects real wall-clock time. If you skip this branch your async traces will all show 0.0ms, which is the kind of bug you only catch by running the decorator under load.