Context Logging for Python

A structured logging library with context propagation for async Python applications. Designed to seamlessly integrate with Nest.js/TypeScript applications using the same header format for context propagation.

Features

• Context Propagation: Context persists across async calls using contextvars
• HTTP Integration: aiohttp middleware and client for HTTP context propagation
• Kafka Integration: aiokafka producer and consumer utilities for Kafka context propagation
• Structured Logging: JSON-formatted logs using structlog
• GCP Cloud Logging: Send logs to Google Cloud Logging with context data
• Sentry Integration: Send errors and warnings to Sentry with context as tags
• TypeScript Compatibility: Uses the same x-sqdgn- header prefix as the TypeScript version

Installation

# Basic installation
pip install context-logging

# With GCP Cloud Logging support
pip install context-logging[gcp]

# With Sentry support
pip install context-logging[sentry]

# With all optional dependencies
pip install context-logging[all]

# For development
pip install context-logging[dev]

Quick Start

Initialize the Logger

from context_logging import initialize_logger, LoggerConfig

config = LoggerConfig(
    app_name="my-service",
    gcp_log_level="INFO",      # Optional: GCP logging level
    sentry_log_level="WARNING", # Optional: Sentry logging level
)
logger = initialize_logger(config)

Using Context

from context_logging import Context, context_provider

# Create a context
ctx = Context({
    "user_id": "123",
    "request_id": "abc-def",
})

# Use the context-bound logger
ctx.logger.info("User logged in", action="login")

# Add more data to context
ctx.merge({"session_id": "xyz"})

# Create a child context
child_ctx = ctx.child({"operation": "fetch_profile"})

HTTP Middleware (aiohttp)

from aiohttp import web
from context_logging import context_middleware, context_provider, initialize_logger, LoggerConfig

# Initialize logger
initialize_logger(LoggerConfig(app_name="my-http-service"))

# Create app with middleware
app = web.Application(middlewares=[context_middleware])

async def handler(request: web.Request) -> web.Response:
    ctx = context_provider.get_context()
    ctx.logger.info("Handling request")
    
    # Context is automatically extracted from x-sqdgn-* headers
    user_id = ctx.get("user_id")
    
    return web.json_response({"ok": True})

app.router.add_get("/api/endpoint", handler)

HTTP Client with Context Propagation

from context_logging import Context, HttpRequestService

async def call_downstream_service():
    ctx = context_provider.get_context()
    
    async with HttpRequestService() as client:
        # Context is automatically added to request headers
        response = await client.get(ctx, "http://other-service/api/data")
        return await response.json()

Kafka Producer with Context

from aiokafka import AIOKafkaProducer
from context_logging import Context, KafkaProducerService

async def send_event():
    producer = AIOKafkaProducer(bootstrap_servers="localhost:9092")
    await producer.start()
    
    service = KafkaProducerService(producer)
    
    ctx = Context({
        "user_id": "123",
        "trace_id": "abc",
    })
    
    # Context is automatically added to message headers
    await service.send_message(
        {"event": "user_action"},
        topics=["events"],
        custom_ctx=ctx,
    )
    
    await producer.stop()

Kafka Consumer with Context Extraction

from aiokafka import AIOKafkaConsumer
from context_logging import KafkaConsumerService, Context

async def handle_message(message: dict, ctx: Context):
    # Context is automatically extracted from message headers
    ctx.logger.info("Processing message", event=message.get("event"))
    
    # Add to context for downstream calls
    ctx.merge({"processing_id": "xyz"})

async def consume_events():
    consumer = AIOKafkaConsumer(
        "events",
        bootstrap_servers="localhost:9092",
        group_id="my-consumer-group",
    )
    await consumer.start()
    
    service = KafkaConsumerService(consumer)
    
    # Messages are processed with automatic context extraction
    await service.consume_forever(handle_message)

Context Propagation

Header Format

Context is propagated using HTTP headers and Kafka message headers with the prefix x-sqdgn-:

x-sqdgn-user_id: 123
x-sqdgn-request_id: abc-def
x-sqdgn-trace_id: xyz

This format is compatible with the TypeScript version of the library.

Automatic Fields

The HTTP middleware automatically adds these context fields:

• req_id: Request ID (from X-Request-Id header or auto-generated)
• req_user_ip: Client IP address
• req_user_agent: User agent string
• http_path: Request path
• http_method: HTTP method

Configuration

Environment Variables

# Logging
LOG_LEVEL=DEBUG          # Default log level (DEBUG, INFO, WARNING, ERROR)

# GCP Cloud Logging
GOOGLE_APPLICATION_CREDENTIALS=/path/to/credentials.json
# OR
GCP_CREDENTIALS_B64=base64_encoded_credentials
GOOGLE_CLOUD_PROJECT=your-project-id

# Sentry
SENTRY_DSN=https://key@sentry.io/project

LoggerConfig Options

@dataclass
class LoggerConfig:
    app_name: str                          # Application name for logs
    location: str = ""                     # Optional location identifier
    node_id: str = ""                      # Optional node identifier
    gcp_log_level: Optional[str] = None    # GCP logging level
    gcp_type_filter: Optional[str] = None  # Filter logs by type
    sentry_log_level: Optional[str] = None # Sentry logging level
    sentry_additional_tags: List[str] = [] # Additional Sentry tags
    json_logs: Optional[bool] = None       # Force JSON output

Testing

Running Unit Tests

# Install dev dependencies
pip install -e ".[dev]"

# Run unit tests
pytest tests/test_context.py -v

Running E2E Tests

# Start Kafka
docker-compose -f docker-compose.e2e.yml up -d

# Configure environment
cp .env.e2e.example .env.e2e
# Edit .env.e2e with your GCP and Sentry credentials

# Run e2e tests
pytest tests/test_e2e_*.py -v

API Reference

Context

class Context:
    def __init__(self, ctx: Optional[Dict] = None)
    def merge(self, ctx: Dict) -> None
    def values() -> Dict[str, str]
    def child(ctx: Optional[Dict] = None) -> Context
    def set(name: str, value: Any) -> None
    def get(name: str) -> Optional[str]
    def for_each(fn: Callable) -> None
    @property
    def logger() -> structlog.BoundLogger

ContextProvider

class ContextProvider:
    @staticmethod
    def get_context() -> Context  # Raises if no context
    @staticmethod
    def get_context_optional() -> Optional[Context]
    @staticmethod
    def set_context(ctx: Context) -> Token
    @staticmethod
    def reset_context(token: Token) -> None
    @staticmethod
    def run(ctx: Context, func: Callable, *args, **kwargs) -> Any
    @staticmethod
    async def run_async(ctx: Context, coro: Coroutine) -> Any

HttpRequestService

class HttpRequestService:
    async def get(ctx: Context, url: str, **kwargs) -> ClientResponse
    async def post(ctx: Context, url: str, data: Any = None, **kwargs) -> ClientResponse
    async def put(ctx: Context, url: str, data: Any = None, **kwargs) -> ClientResponse
    async def delete(ctx: Context, url: str, **kwargs) -> ClientResponse
    async def patch(ctx: Context, url: str, data: Any = None, **kwargs) -> ClientResponse

KafkaProducerService

class KafkaProducerService:
    async def send_message(
        message: Any,
        topics: List[str],
        custom_ctx: Optional[Context] = None,
        headers: Optional[Dict] = None,
        **kwargs,
    ) -> None

KafkaConsumerService

class KafkaConsumerService:
    async def consume_forever(handler: Callable[[Any, Context], Awaitable[None]]) -> None
    async def consume_one(handler: Callable, timeout_ms: int = 1000) -> Optional[Any]
    async def consume_batch(handler: Callable, max_records: int = 100) -> int

License

MIT License