Metadata-Version: 2.4
Name: logging-mixin
Version: 0.3.0
Summary: Class-bound structured logging with auto-injected correlation IDs for Python services.
Project-URL: Homepage, https://github.com/jekhator/logging-mixin
Project-URL: Repository, https://github.com/jekhator/logging-mixin.git
Project-URL: Issues, https://github.com/jekhator/logging-mixin/issues
Project-URL: Changelog, https://github.com/jekhator/logging-mixin/releases
Author: C. James Ekhator
License: Apache-2.0
License-File: LICENSE
Keywords: correlation-id,distributed-tracing,logging,observability,structured-logging
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Logging
Classifier: Typing :: Typed
Requires-Python: >=3.11
Provides-Extra: all
Requires-Dist: botocore; extra == 'all'
Requires-Dist: celery; extra == 'all'
Requires-Dist: httpx; extra == 'all'
Requires-Dist: requests; extra == 'all'
Provides-Extra: botocore
Requires-Dist: botocore; extra == 'botocore'
Provides-Extra: celery
Requires-Dist: celery; extra == 'celery'
Provides-Extra: dev
Requires-Dist: botocore; extra == 'dev'
Requires-Dist: celery; extra == 'dev'
Requires-Dist: httpx; extra == 'dev'
Requires-Dist: mypy; extra == 'dev'
Requires-Dist: pytest-asyncio; extra == 'dev'
Requires-Dist: pytest-cov; extra == 'dev'
Requires-Dist: pytest>=8; extra == 'dev'
Requires-Dist: requests; extra == 'dev'
Requires-Dist: types-requests; extra == 'dev'
Provides-Extra: httpx
Requires-Dist: httpx; extra == 'httpx'
Provides-Extra: requests
Requires-Dist: requests; extra == 'requests'
Description-Content-Type: text/markdown

# logging-mixin

[![PyPI version](https://img.shields.io/pypi/v/logging-mixin.svg)](https://pypi.org/project/logging-mixin/)
[![CI](https://github.com/jekhator/logging-mixin/actions/workflows/ci.yml/badge.svg)](https://github.com/jekhator/logging-mixin/actions/workflows/ci.yml)
[![License](https://img.shields.io/badge/license-Apache%202.0-blue.svg)](LICENSE)
[![Python Versions](https://img.shields.io/pypi/pyversions/logging-mixin.svg)](https://pypi.org/project/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:

```python
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

```bash
uv add logging-mixin
```

Or with pip:

```bash
pip install logging-mixin
```

With optional dependencies for specific frameworks/clients:

```bash
# 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:

```bash
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:

```python
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

```python
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:

```python
from logging_mixin.adapters.asgi.asgi_client import CorrelationIdMiddleware

app.add_middleware(CorrelationIdMiddleware)
```

### 3. Use LoggingMixin in your classes

```python
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:

```python
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):

```python
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:

```python
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)

```python
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

```python
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:

```python
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

```python
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.
