YAPL Kit

Yet Another Python Logger

A modern, minimal, and extensible logging toolkit built on top of Python's standard logging module.

YAPL keeps the stdlib logging ecosystem intact while adding structured logging, queue-backed delivery, formatting, coloring, buffering, stack helpers, and webhook support.

All YAPL feature families ship in the single yapl-kit distribution. The base install uses only the standard library; optional integrations install their runtime dependencies through extras.

Installation

YAPL Kit requires Python 3.13 or newer.

With uv:

uv add yapl-kit

With pip:

pip install yapl-kit

Install webhook or OpenTelemetry integrations only when needed:

pip install 'yapl-kit[webhook]'
pip install 'yapl-kit[opentelemetry]'

With uv, use uv add 'yapl-kit[webhook]' or uv add 'yapl-kit[opentelemetry]'.

The canonical import namespace is yapl:

import yapl

The package also exposes a root-level compatibility namespace:

import yapl_kit

Documentation examples use yapl.

Quick Start

import yapl

yapl.setup_logging(
    level="INFO",
    outputs=("stdout",),
    format="text",
)

log = yapl.get_logger(__name__)

log.info("Hello world")
log.warning("Something happened")

Structured fields can be attached directly to log calls:

log.info("user_login", user_id=123, ip="1.2.3.4")
log.error("payment_failed", order_id="A-17", exc_info=True)

Shared request or job fields can be scoped to a block:

with yapl.context(request_id="req-42", user_id=123):
    log.info("request_started")
    log.info("request_finished")

For orderly application shutdown, call:

yapl.shutdown_logging()

This gives queued work its configured bounded drain window. Records still pending when the shutdown deadline expires may be discarded.

Use yapl.flush_logging() to drain queued work while keeping logging active, or use yapl.logging_session(...) for scripts that should configure and shut down YAPL around one block. yapl.logging_status() exposes queue and webhook delivery counters for operational checks.

logging_session() owns the process-wide YAPL configuration for its block and does not restore an earlier YAPL setup; use it at an application or script boundary rather than nesting it inside another logging session.

Design

YAPL does not replace Python logging. It builds on stdlib handlers, formatters, filters, LogRecord objects, and propagation.

The main design goals are:

• keep the distribution small and focused
• preserve compatibility with Python's logging ecosystem
• keep expensive formatting and I/O off the application hot path
• make queueing, retries, and shutdown behavior bounded
• keep optional features isolated and explicit
• remain easy to understand and maintain

Configuration

YAPL supports direct setup arguments as well as typed configuration. Install yapl-kit[webhook] before configuring a webhook sink.

from yapl import WebhookConfig, YaplConfig, get_logger, setup_logging

config = YaplConfig(
    queue_max_size=2_000,
    webhook=WebhookConfig(
        "https://example.com/webhook",
        min_level="ERROR",
    ),
)

setup_logging(
    config,
    outputs=("stdout", "webhook"),
)

log = get_logger(__name__)
log.info("Application started")

Setup arguments override the supplied YaplConfig for that call without mutating the configuration object.

Filtering and custom levels

Use suppression rules to drop matching records from every configured sink. All conditions in one rule must match; a record is dropped when any rule matches. Use an OutputConfig filter when a condition should apply to just that sink. Sink filters can be standard logging.Filter objects or callables that accept a LogRecord and return whether to keep it.

import logging

import yapl
from yapl import OutputConfig, SuppressionRules, YaplConfig

NOTICE_LEVEL = 25
yapl.register_level("NOTICE", NOTICE_LEVEL)
yapl.register_level_method("NOTICE")

config = YaplConfig(
    level=logging.DEBUG,
    queue_enabled=False,
    sinks=(
        OutputConfig(
            "stdout",
            filters=(lambda record: record.levelno >= NOTICE_LEVEL,),
        ),
    ),
    suppression=SuppressionRules().drop(
        logger="myapp.noisy",
        logger_prefix=True,
    ),
)

try:
    log = yapl.setup_logging(config)
    log.notice("operator_notice")
finally:
    yapl.shutdown_logging()

Register a custom level before using it in configuration or logging calls. The optional register_level_method call adds a convenience method such as log.notice; without it, use the standard log.log(NOTICE_LEVEL, ...) method. The root logging level and each sink's filter are separate gates: this example keeps DEBUG enabled globally, then applies the NOTICE threshold only to the console sink. See development/demos/filters_and_levels.py for a runnable example with both suppression rules and sink filtering.

Output Profiles

A single process can use different output profiles for different sinks.

import yapl
from yapl import OutputConfig, YaplConfig

config = YaplConfig(
    sinks=(
        OutputConfig(
            "stdout",
            format="text",
            color="auto",
            theme_name="dark",
        ),
        OutputConfig(
            "file",
            format="jsonl",
            file_path="service.jsonl",
        ),
    ),
    resource={
        "service.name": "billing",
        "deployment.environment": "production",
    },
)

log = yapl.setup_logging(config)
log.info("payment_processed", request_id="req-42")

Use resource for stable source metadata such as service, environment, region, or cluster. Keep request IDs, user IDs, trace IDs, and other high-cardinality values as normal structured fields. File output profiles also support bounded rotation with file_max_bytes and file_backup_count; those options are valid only for file sinks.

Trace correlation

When an application uses OpenTelemetry, YAPL can copy the active span's IDs onto each record so structured outputs include trace_id and span_id:

from yapl import YaplConfig, setup_logging, shutdown_logging

setup_logging(YaplConfig(capture_trace_context=True))
# Call this during application shutdown to drain queued output.
shutdown_logging()

Install the opentelemetry extra, which supplies opentelemetry-api, and configure an OpenTelemetry SDK or instrumentation package to create active spans. YAPL reads the IDs while the log call runs, so they are retained when sink work is queued. Existing trace_id or span_id record fields take precedence.

Per-sink processing

Each OutputConfig can add processing that applies only to that sink:

import logging

from yapl import OutputConfig, YaplConfig
from yapl.extensions.formatting import AddServiceName, RedactSecrets


class OnlyErrors(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        return record.levelno >= logging.ERROR


config = YaplConfig(
    sinks=(
        OutputConfig(
            "stdout",
            format="text",
            processors=(AddServiceName("checkout-console"),),
            enrichment={"audience": "operator"},
            filters=(OnlyErrors(),),
        ),
        OutputConfig(
            "file",
            format="jsonl",
            file_path="checkout.jsonl",
            processors=(RedactSecrets(("password", "token")),),
            fields=("timestamp", "level", "message", "context"),
        ),
    ),
)

filters are standard record filters and enrichment adds context fields on an isolated copy of the record. processors transform the normalized event. redact_secret_fields and redaction_replacement provide a concise redaction policy, schema maps output names to dotted event paths, and fields retains top-level event fields. A sink with a stricter redaction policy cannot leak its unredacted record through another sink: each sink receives an isolated record and event copy.

The deterministic order is: record filters, context enrichment, global processors, sink processors, redaction, schema projection, field selection, then formatting or webhook payload construction. Existing per-sink levels and formatters still run at their normal handler boundaries. Queue workers apply the same policy before each target handler, and webhook processing remains on the webhook worker.

RedactSecrets remains available for custom processor pipelines. When using redact_secret_fields, set it to an empty tuple to explicitly disable the webhook's default redaction for that sink; omit it to retain the webhook configuration's default policy.

For a standard-library custom formatter, set OutputConfig.formatter. YAPL applies that sink's processors and event policy to an isolated record before delegating to the formatter. The transformed normalized event is also available as record.yapl_event; without sink policies, the original formatter instance is used unchanged.

Formatting

Structured, text, JSON Lines, and logfmt output are available through the formatting extension.

from yapl.extensions.formatting import (
    JSONFormatter,
    LogfmtFormatter,
    TextFormatter,
)

formatter = JSONFormatter()

Color output is restricted to human-facing text/plain terminal sinks. JSON, JSONL, logfmt, files, and webhooks remain machine-readable.

color="auto" emits ANSI only for a compatible terminal and respects NO_COLOR.

Queueing

Queue-backed handlers keep sink work away from application threads.

from yapl.extensions.queueing import QueueingConfig, QueueingHandler

In the default queued configuration:

• queue insertion is bounded and non-blocking
• formatting runs on workers
• file and console sink work runs on workers
• webhook payload construction and network I/O run on workers
• retries and retry delays are bounded
• shutdown waiting is bounded

Queue capacity limits the number of retained records, not their total byte size. Queued records may retain message arguments, extra values, exceptions, tracebacks, and object graphs until processed or discarded.

Webhooks

Webhook delivery requires the optional webhook extra, which installs httpx:

pip install 'yapl-kit[webhook]'

Configure a webhook:

import yapl
from yapl.extensions.webhook import WebhookConfig

webhook = WebhookConfig(
    url="https://example.com/webhook",
    min_level="ERROR",
)

log = yapl.setup_logging(webhook=webhook)

Webhook work runs in the background so application threads do not intentionally wait for network I/O. Active webhook handlers reuse their HTTP client, and timeout/retry behavior is bounded.

Buffering

Buffering captures records during a block and then replays or discards them when the block exits.

from yapl.extensions.buffering import (
    BufferedLogs,
    BufferingConfig,
    install_buffering,
)

install_buffering()

with BufferedLogs(BufferingConfig(flush_mode="ordered")):
    log.info("preflight", phase="warmup")
    log.warning("preflight_warning", stage=1)

The default behavior flushes buffered records on success and on error. Configuration can be used to discard them instead when a block should remain silent.

Extended Levels

YAPL can register extra development levels:

import yapl

yapl.setup_logging(level_mode="extended")

log = yapl.get_logger(__name__)
log.trace("trace message")
log.dev("dev message")

Built-in extended levels:

Level	Value
TRACE	5
DEV	15

Stack Helpers

Callsite and tracing helpers live under yapl.stack.

from yapl import get_logger, setup_logging
from yapl.stack import FunctionCallTracer, get_callsite

setup_logging(level_mode="extended")
callsite = get_callsite()

tracer = FunctionCallTracer(get_logger(__name__))
tracer.enable_current_thread()
try:
    run_complex_operation()
finally:
    tracer.disable_current_thread()

Guarantees and Non-Guarantees

In the default queued configuration, remote and sink I/O never runs on the caller thread, and queue insertion does not intentionally wait.

This is an operational guarantee for YAPL's queue and built-in webhook path, not an absolute claim that logging can never block under every possible configuration.

Guaranteed by default

• Queue-backed delivery has a bounded record count.
• Queue insertion uses non-blocking insertion under the default overflow policy.
• Formatting, file/console delivery, webhook payload work, HTTP, DNS, and retry waits happen on background workers.
• Webhook retry attempts and retry delays are bounded.
• Active webhook handlers reuse a lazily created HTTP client.
• Shutdown waiting is bounded by shutdown_timeout_seconds.
• Records may be discarded when the shutdown deadline expires so the caller can continue.

Not guaranteed

• Delivery of every record during overload, remote failure, or shutdown.
• A fixed byte-level memory limit.
• Termination of arbitrary custom handlers already blocked forever.

Python cannot safely terminate arbitrary user handler code. Keep custom handlers non-blocking or give them their own timeouts.

Explicit opt-outs

overflow_policy="block" deliberately enables application-level backpressure and may block the caller. With enqueue_timeout_seconds=None, that wait may be indefinite.

queue_enabled=False runs normal stdlib handlers directly, so file, console, or custom sink work may occur on the caller thread.

Extension Modules

YAPL keeps feature families under explicit modules within the main distribution:

from yapl.stack import FunctionCallTracer, get_callsite
from yapl.extensions.buffering import BufferedLogs, BufferingConfig
from yapl.extensions.coloring import install_color_structured_console
from yapl.extensions.formatting import JSONFormatter, StructuredFormatterConfig
from yapl.extensions.queueing import QueueingConfig
from yapl.extensions.webhook import WebhookConfig

The yapl.extensions package contains the concrete feature implementations.

Package Philosophy

YAPL is intended to be a small quality-of-life logging toolkit rather than a replacement observability framework.

The focus is on:

• predictable behavior
• low caller-side overhead
• bounded background work
• readable configuration
• stdlib compatibility
• optional features without unnecessary runtime weight