without_logging¶
A without pipeline for logs: stdlib log records parsed into immutable values, filtered and enriched as processors, drained to a sink.
CaptureHandler(
loop: AbstractEventLoop,
queue: Queue[Record],
parse: Callable[[LogRecord], Record],
)
Bases: Handler
A stdlib handler that parses each LogRecord and offers it to an asyncio queue.
This is the one-way bridge at the ingestion edge. Stdlib logging is the
de-facto narrow waist every third-party library already writes to, so making
it a source (rather than fighting it) is the whole trick: a pushed
LogRecord is parsed to a Record here and dropped onto the queue that
stream_from_queue turns into the pull-based stream the pipeline consumes.
Control only ever flows outward from stdlib into without, never back.
A log call MAY happen off the event loop thread, so the parsed value is handed
to the loop with call_soon_threadsafe. The queue is bounded, so a burst that
outruns the sink is dropped rather than growing memory without limit;
dropped counts how many so an operator can see it. The overflow policy is a
boundary decision the app owns: raise capacity, or pass capacity=None for
an unbounded queue that never drops (at the cost of the bound).
Bases: IntEnum
The five standard severities as an ordered value, for writing filters.
Members are the stdlib numeric levels, so a Record.level (a plain int)
can be compared straight against them: record.level >= Level.WARNING.
Keeping Record.level an int rather than this enum is deliberate: a
third-party library that logs at a non-standard numeric level is carried
through as that number, not rejected, so the enum is a vocabulary for
predicates rather than a closed set the record must belong to.
dataclass
¶
Record(
timestamp: datetime,
level: int,
logger: str,
message: str,
exception: TracebackException | None,
fields: Mapping[str, object],
)
One log event as an immutable value, parsed from a stdlib LogRecord.
Where a LogRecord is a mutable bag that formatters and filters rewrite in
place, a Record is a value: it always means the same thing, so it can be
filtered, enriched, fanned out, and sunk without any stage disturbing
another's copy. Enrichment returns a new record (with_fields) rather than
mutating this one. fields is the record's structured data, exposed read-only:
seeded from the log call's extra= at the parse edge, then grown by
enrichment (add_fields, a bind merged by merge_context, any with_fields).
It is named for what it holds, not for extra= alone, because those later
sources write here too. exception is a TracebackException captured at the
ingestion edge (None when the event carried none): a structured value,
not a rendered string, so how the traceback is formatted stays a downstream
boundary the app owns ("".join(exc.format()) for the stdlib text, or walk
exc.stack for JSON frames), the same way rendering the record itself is.
Extracting it drops all references to the live frames, so the record stays a
self-contained value with no traceback pinned to it.
Bind fields onto every record logged within this block, on this task.
The call-site half of structured context, the equivalent of structlog's
bind_contextvars: with bind(request_id=...) stamps those fields onto every
record logged inside the block. Scoping is explicit and lexical: the fields
are in effect for the block and gone after it, leaving no shared place
mutated. It writes a task-local ContextVar, so concurrent tasks each see
only their own binds, and nested binds accumulate (an inner bind merges onto
the outer and is undone on exit; a key set by both takes the inner value
inside the inner block).
Binding alone does nothing until the capture side reads it: pair this with
merge_context, which merges the bound fields onto each record at the parse
edge, where they are still live (the handler's emit runs synchronously on
the logging call's task). A downstream processor cannot read them, having left
the caller's context for the sink task.
Merge the context bound by bind onto a record: the read half of call-site context.
A Record -> Record enrichment, composed on top of a parser rather than
wrapping it: capture's default parser applies it after parse_record, and a
custom parser composes it the same way,
merge_context(parse_record(log_record)). A field the log call set explicitly
via extra= wins over a bound field of the same name (the more specific
value), so bind supplies defaults, never overrides a per-call value.
It reads the task-local ContextVar that bind writes, so it MUST run at the
parse edge, in the handler's emit on the caller's task, not as a pipeline
Processor: the pipeline runs in the sink task, having left that context,
where the bind would read as empty.
A processor that merges static fields onto every record: enrichment.
Enrichment is expressible as a from_map (one record in, one enriched
record out), so it uses the core builder directly. Enriching from a shared
behavior (a value the whole pipeline sees the latest of: the current config
revision, a sampling rate) is the same shape, reading it via current()
inside the step. Per-call-site context (a request or trace id, which is
task-local) is not recoverable here: the pipeline runs in the sink task,
having left the caller's context. Bind it at the edge instead with
bind/merge_context, which read it while emit is still on the caller's
task (see the guide).
A predicate matching records at level or more severe: the level threshold.
Pair it with the core from_selector to keep only those records (or
from_filter to drop them). It is async to match the one color of
predicate from_selector/from_filter expect, though the decision itself
just compares integers. Because a Record.level is a plain int, a Level
member reads naturally as the argument (at_least(Level.WARNING)), but any
numeric level works.
Parse a mutable stdlib LogRecord into an immutable Record value.
The ingestion boundary: this is where a pushed LogRecord, from the app's
own calls or any third-party library's, becomes the typed value the rest of
the pipeline works with. It is a pure function (LogRecord -> Record), so the
whole translation is testable without touching the logging machinery. The
structured fields are every attribute on the record outside the standard
envelope (RESERVED), which is exactly what logging's extra= merges in;
the message is rendered and any exception is captured as a structured value
here (see extract_exception) so nothing live is carried past this point.
exception_to_dict(
exception: TracebackException,
) -> dict[str, object]
Convert a captured TracebackException into a JSON-serializable dict: structured traceback.
Each frame becomes {file, line, function, code} (from the extracted
FrameSummary values, no live frames), and a chained exception is nested under
cause, following Python's own rule: an explicit raise ... from (__cause__)
takes precedence, else the implicit __context__ unless it was suppressed. This
is the read half render_json uses, exposed so a custom JSON renderer can reuse
it.
exception_to_text(exception: TracebackException) -> str
Render a captured TracebackException to its full traceback text.
stdlib's own multi-line rendering (format), chained cause/context and all,
as a single string. The flat-string alternative to the structured
exception_to_dict when encoding a record's exception, and what
render_console uses.
The default timestamp format for the renderers: ISO 8601 (datetime.isoformat).
A Record -> str renderer emitting one human-readable line per record.
TIMESTAMP LEVEL logger "message" {key=value, ...}: the message is quoted and
the fields grouped in braces (each value repr'd), so the free-text message and
the structured fields never blur into each other or the leading metadata. The
braces are omitted when there are no fields. An exception is appended as its full
traceback text (exception_to_text, cause chain and all), indented, on following
lines. timestamp chooses the time format (default iso_timestamp). No coloring:
wrap the line, or write your own from_map(Record -> str), if you want ANSI.
Optional and opt-in, like render_json: compose(render_console(),
offload(to_stream(sys.stderr))).
render_json(
*,
timestamp: Callable[[datetime], object] = iso_timestamp,
exception: Callable[
[TracebackException], object
] = exception_to_dict,
) -> Processor[Record, str]
A Record -> str renderer emitting one JSON object per record: structured logging.
The envelope (timestamp ISO-8601, level, logger, message) and every
field flat at the top level, so a log aggregator can index them directly; the
envelope wins a name clash, so a field cannot shadow level or message.
timestamp chooses how the time is encoded (default iso_timestamp; pass, say,
a lambda when: when.timestamp() for an epoch number). exception chooses how a
traceback is encoded: exception_to_dict (default, structured frames, stays
queryable) or exception_to_text (the flat traceback string), or any
TracebackException -> <json value> of your own. A
non-serializable field value is coerced with str (default=str): the log
JSON is machine-consumed, so a clean indexable value beats a Python repr, and
str still falls back to the repr form for an object without its own
__str__ (render_console uses repr, matching its human-scanning medium). A
stray value therefore never tears down the log pipeline.
Optional and opt-in: the core ships no mandatory formatter (encoding is the app's
boundary), so compose this before a string sink yourself, e.g.
compose(render_json(), offload(to_stream(sys.stdout))).
A schedule for to_rotating_file: the next occurrence of any of these wall-clock times.
Given a moment, returns the soonest of times strictly after it, interpreted in
timezone tz. So at_times(time(0, 0)) rotates daily at midnight tz, and
at_times(time(0, 0), time(12, 0)) twice a day. The list of times is the
recurring cycle of rotation boundaries, resolved to a concrete datetime against
the actual current time each call, so nothing goes stale and a boundary missed
while idle collapses to a single rotation. Strictly-after (not at-or-after) is
what lets the writer advance past a boundary it just hit without looping on it.
async
¶
offload(
work: Callable[[Iterator[list[T]]], None],
) -> AsyncIterator[Sink[T]]
Run a blocking work on a dedicated thread, fed by the yielded async Sink.
The point is efficient blocking I/O under heavy logging. An async file library
(aiofiles, anyio) hops to a worker thread per operation, so a busy log
stream pays that round-trip on every write. Here a single long-lived thread
owns the resource and does all the I/O: work is plain blocking Python that
consumes the items, and the yielded Sink just drops each onto a thread-safe
queue the worker drains. No per-item thread hop, and the "processor body" stays
ordinary synchronous code.
Items arrive in bursts: each element of the iterator is everything available on the queue at that instant (at least one item, blocking for the first). A burst boundary is therefore exactly the moment the worker has caught up, which is where a writer flushes: under load bursts are large and flushes are few, and when idle each burst is a single item flushed at once. So durability needs no flush-frequency knob; it falls out of the queue's own backlog.
Lifecycle is bounded by the with block: the thread starts on entry, and on
exit the queue is shut down so the worker drains what remains and then ends,
and the thread is joined (so a file is closed before the block returns). If
work raises, that surfaces when the block exits.
Nest this outside the consumer that drives the sink, so the worker outlives
the draining. With capture, that means async with offload(...) as writer:
around async with capture(..., writer):, not the other way round.
The queue is unbounded in this first cut: the async side never blocks or drops,
at the cost of growing memory if the worker cannot keep up with a sustained
burst (a stalled disk). A bounded, drop-counting variant is a deliberate
follow-up. The bidirectional case (a full Processor bridged onto a thread,
with an output queue as well) is intentionally out of scope; this covers the
terminal-sink need (writing) without that extra state.
to_rotating_file(
name: Callable[[int, datetime], Path],
*,
max_bytes: int | None = None,
max_age: timedelta | None = None,
schedule: Callable[[datetime], datetime] | None = None,
now: Callable[[], datetime] = now_utc,
) -> Callable[[Iterator[list[str]]], None]
A blocking worker (for offload) that appends lines to a file, rotating by size and/or time.
Rotation lives in the worker deliberately: it is the one place with all the
information to decide, timed with its own writes. Size is a function of the
bytes written, so a size limit cannot be an independent, decoupled trigger; the
worker keeps an exact byte count (writing UTF-8 bytes directly) rather than
polling a lagging on-disk size, and reads the clock the same way. So max_bytes
(size), max_age (elapsed since the file opened, a relative interval),
schedule (the next absolute wall-clock boundary, e.g. midnight, via a
next-boundary function such as at_times), and any combination (rotate on
whichever trips first) all work here, which the stdlib's separate size and time
handlers cannot do at once. At least one policy is required: with no limit the
file would never rotate (an unbounded file), which is not the affordance the
name suggests, so that case raises ValueError rather than being silently
allowed.
schedule(t) returns the next rotation boundary strictly after t; the writer
samples it at each open and rotates once now() reaches it (missed boundaries
collapse to one rotation). It is the pure, thread-synchronous form of "a stream
of rotation datetimes": the worker already owns the clock, so it generates the
stream itself rather than sampling an async one.
name turns a rotation index and the rotation time into a path
(lambda i, when: directory / f"app.{i}.log"); index 0 is the initial file,
appended to if it already exists, and the time is that file's open time drawn
from the same injected clock, so timestamped names stay consistent with the
rotation decision. now is the clock, injected so time-based rotation is
deterministic under test. This writes strings
(render a Record to text in a from_map in front), owns the newline framing,
and flushes at each burst boundary. The rotation decision is made before each
write (so a size limit is not overshot, bar a single line larger than
max_bytes), and it is inlined and short-circuiting: now() is read only when a
size check has not already forced the rotation and an age limit is set.
A blocking worker (for offload) that appends each string as a line to an open text stream.
The destination-shaped sibling of to_rotating_file, for a stream the caller
already owns: sys.stderr, a socket's text wrapper, an in-memory buffer. It
writes strings (render a Record to text with a from_map in front), owns the
newline framing, and flushes at each burst boundary. Unlike to_rotating_file it
does not open or close the stream: the caller's stream outlives this and may be
shared, so closing it (sys.stderr, say) is not this worker's to do.