| """Structured logging — structlog with JSON in prod, console in dev. |
| |
| Spec: docs/Specs.md §15, docs/14-Engineering.md §3.1 |
| Every log carries `correlation_id` when present in context. |
| """ |
|
|
| import logging |
| import sys |
| from typing import Any |
|
|
| import structlog |
|
|
|
|
| def configure_logging(*, level: str = "INFO", env: str = "development") -> None: |
| """Idempotent. Safe to call from app startup and from tests.""" |
|
|
| logging.basicConfig( |
| format="%(message)s", |
| stream=sys.stdout, |
| level=getattr(logging, level), |
| ) |
|
|
| shared_processors: list[Any] = [ |
| structlog.contextvars.merge_contextvars, |
| structlog.processors.add_log_level, |
| structlog.processors.TimeStamper(fmt="iso", utc=True), |
| structlog.processors.StackInfoRenderer(), |
| structlog.dev.set_exc_info, |
| ] |
|
|
| if env == "development": |
| renderer: Any = structlog.dev.ConsoleRenderer(colors=True) |
| else: |
| renderer = structlog.processors.JSONRenderer() |
|
|
| structlog.configure( |
| processors=[*shared_processors, renderer], |
| wrapper_class=structlog.make_filtering_bound_logger(getattr(logging, level)), |
| context_class=dict, |
| logger_factory=structlog.PrintLoggerFactory(), |
| cache_logger_on_first_use=True, |
| ) |
|
|
|
|
| def get_logger(name: str | None = None) -> structlog.stdlib.BoundLogger: |
| return structlog.get_logger(name) |
|
|
