logging-mixin 0.3.0

Class-bound structured logging with auto-injected correlation IDs for Python services.

logging-mixin

End-to-end correlation-ID propagation for Python services. Automatic correlation-ID injection across logs, HTTP clients, task queues, and AWS services. Built for distributed systems.

• Correlation-ID context via contextvars.ContextVar — survives async/await, thread pools, and background tasks
• 8 adapters for inbound/outbound/task/logging/cloud scenarios
• LoggingMixin class and logged decorator for zero-boilerplate logging
• Python 3.11+ with uv package management

What It Does

Tracking a single request through a distributed system requires correlation IDs on every log line, HTTP call, database query, and background task. Traditional approaches require threading the ID through every function.

logging-mixin propagates correlation IDs automatically:

from logging_mixin import LoggingMixin, set_correlation_id

# Request handler: set once
def handle_request(request):
    set_correlation_id(request.headers.get("X-Correlation-ID", "req-123"))
    service = OrderService()
    service.create_order(123)  # Correlation ID is now in the context

# Service: logs include correlation ID automatically
class OrderService(LoggingMixin):
    def create_order(self, user_id: int):
        self.log_info("order.create", user_id=user_id)
        # Logs with: {"correlation_id": "req-123", "user_id": 123, ...}
        
        # Outbound HTTP call: correlation ID injected automatically
        self.send_notification(user_id)

# Background task: inherits correlation ID from request context
@celery.shared_task
def send_notification(user_id: int):
    self = NotificationService()
    self.log_info("notification.send", user_id=user_id)
    # Same correlation ID propagates here

Install

uv add logging-mixin

Or with pip:

pip install logging-mixin

With optional dependencies for specific frameworks/clients:

# Individual adapters
uv add "logging-mixin[httpx]"       # HTTPX client instrumentation
uv add "logging-mixin[requests]"    # Requests client instrumentation
uv add "logging-mixin[celery]"      # Celery task propagation
uv add "logging-mixin[botocore]"    # AWS SDK instrumentation

# Install all adapters at once
uv add "logging-mixin[all]"

Or with pip:

pip install "logging-mixin[all]"

Requires Python 3.11+ (3.11 and 3.12 tested).

Quick Start

1. Add stdlib adapter to your logging config

Stamps correlation_id on every log record:

import logging
from logging_mixin.adapters.stdlib.stdlib_client import CorrelationLogFilter

# Add the filter to your logger
logging.basicConfig()
logging.getLogger().addFilter(CorrelationLogFilter())

2. Set correlation ID at request boundary

from logging_mixin import set_correlation_id

# FastAPI
from fastapi import FastAPI, Request
app = FastAPI()

@app.middleware("http")
async def correlation_middleware(request: Request, call_next):
    set_correlation_id(request.headers.get("X-Correlation-ID", str(uuid.uuid4())))
    return await call_next(request)

Or use the built-in ASGI adapter:

from logging_mixin.adapters.asgi.asgi_client import CorrelationIdMiddleware

app.add_middleware(CorrelationIdMiddleware)

3. Use LoggingMixin in your classes

from logging_mixin import LoggingMixin

class UserService(LoggingMixin):
    def create_user(self, name: str):
        self.log_info("user.create", name=name)
        # Logs include correlation_id automatically
        self.save(name)

4. Instrument outbound clients

HTTP clients automatically inject X-Correlation-ID header:

from logging_mixin.adapters.httpx.httpx_client import CorrelationIdInjector
from logging_mixin.adapters.requests.requests_client import CorrelationHTTPAdapter

# For httpx
client = httpx.Client(event_hooks=CorrelationIdInjector.event_hooks())
client.get("https://api.example.com")  # Sends X-Correlation-ID header

# For requests
session = requests.Session()
CorrelationHTTPAdapter.register_on_session(session)
session.get("https://api.example.com")  # Sends X-Correlation-ID header

AWS SDK (botocore):

from logging_mixin.adapters.botocore.botocore_client import CorrelationIdInjector

s3 = boto3.client("s3")
CorrelationIdInjector.register_on_client(s3)
s3.get_object(Bucket="my-bucket", Key="file.txt")  # Includes correlation ID in AWS service calls

5. Propagate to background tasks (Celery)

Install the optional [celery] dependency, then:

from celery import Celery
from logging_mixin.adapters.celery.celery_client import CorrelationSignals

app = Celery()
CorrelationSignals.connect()

@app.task
def process_order(order_id: int):
    service = OrderService()
    service.log_info("processing", order_id=order_id)
    # Correlation ID from the original request is automatically here

Core API

LoggingMixin (instance methods)

from logging_mixin import LoggingMixin

class MyService(LoggingMixin):
    def do_something(self):
        self.log_debug("debug message", key="value")        # DEBUG level
        self.log_info("info message", key="value")          # INFO level
        self.log_warning("warning message", key="value")    # WARNING level
        self.log_error("error message", key="value")        # ERROR level
        self.log_exception("error with traceback")          # ERROR level + traceback

All methods:

• Accept an event name (string) + optional keyword arguments
• Automatically inject correlation_id into log extra dict
• Read from the per-class logger (module.ClassName)
• Support composition with masking mixins (call mask_for_logging() if it exists)

Correlation context

from logging_mixin import get_correlation_id, set_correlation_id, clear_correlation_id

cid = get_correlation_id()            # Get current correlation ID (None if not set)
set_correlation_id("my-request-id")   # Set manually (tests, background tasks)
clear_correlation_id()                # Clear (test isolation, request boundaries)

Logged decorator

Decorate LoggingMixin methods to auto-log entry/exit and errors:

from logging_mixin import LoggingMixin, logged

class StripeClient(LoggingMixin):
    @logged("stripe.create_intent")
    def create_intent(self, customer_id: str) -> dict:
        return {"status": "ok"}

# Logs "stripe.create_intent.start" on entry
# Logs "stripe.create_intent.error" on exception (with error_type and code)

See docs/apps/decorators/logged.md for detailed usage and composability with @phi_aware and @translate decorators.

The 8 Adapters

All adapters live in logging_mixin/adapters/:

Inbound HTTP (request entry points)

• ASGI (asgi.py) — FastAPI, Starlette, Quart, async WSGI. Extract correlation ID from request headers or generate UUID. Inject into response headers. Includes security hardening (CRLF injection, log injection, DoS protection).
• WSGI (wsgi.py) — Django, Flask, Pyramid, synchronous frameworks. Extract/generate correlation ID. Inject response header. Works alongside ASGI if needed.

Outbound clients (propagate downstream)

• HTTPX (httpx.py) — httpx.Client and httpx.AsyncClient. Instrument to inject X-Correlation-ID header on every request.
• Requests (requests.py) — requests.Session. Instrument via HTTP adapter to inject X-Correlation-ID on every request.
• Botocore (botocore.py) — AWS SDK (boto3). Hook into botocore event system to inject correlation ID into AWS service calls.

Task/async boundaries

• Celery (celery.py) — Propagate correlation ID across task publish → prerun → postrun via Celery signals. Works with all task types (sync, async, delayed).

Logging

• Stdlib (stdlib.py) — logging.Filter. Stamps correlation_id on every LogRecord automatically. Minimal setup.

Cloud/serverless

• Cloud (cloud.py) — AWS Lambda event extraction. Supports API Gateway (v1/v2), ALB, SQS, SNS, EventBridge, and direct-invoke. Auto-generate fallback correlation ID if not present in the event.

See docs/apps/adapters/ for detailed per-adapter documentation.

Design Principles

• ContextVar-based — Survives async/await, thread pools, and background tasks
• Instance-only — LoggingMixin methods read self._logger (cannot be used in @classmethod/@staticmethod)
• Framework-agnostic — Core library has zero dependencies
• Adapter ecosystem — Install only what you use (celery, requests, etc. are optional)
• Security-hardened — ASGI/WSGI adapters validate all input (CRLF injection, control characters, length limits)
• Composable — Works with masking mixins and other decorators

Testing

import logging
from logging_mixin import LoggingMixin, set_correlation_id

def test_logs_with_correlation_id(caplog):
    set_correlation_id("test-123")
    
    service = MyService()
    with caplog.at_level(logging.INFO):
        service.do_something()
    
    assert caplog.records[0].correlation_id == "test-123"

Design Trade-offs

• @classmethod/@staticmethod — LoggingMixin cannot be used there (use module logger + manual injection)
• Implicit behavior — Correlation ID is silently injected (can be surprising if not documented)
• Setup required — Must call set_correlation_id() at request entry or use a framework adapter

License

Apache 2.0 — see LICENSE file.