tiphys.logging

  1"""
  2Structured logging configuration for Tiphys.
  3
  4Uses structlog for structured JSON logging with request-scoped context propagation.
  5
  6Features:
  7- Request-scoped logging context (correlation IDs, channel, peer info)
  8- Context propagation through async operations
  9- Structured JSON output for production
 10- Colored console output for development
 11"""
 12
 13from __future__ import annotations
 14
 15import logging
 16import sys
 17from collections.abc import MutableMapping
 18from contextvars import ContextVar
 19from dataclasses import dataclass, field
 20from typing import Any
 21from uuid import uuid4
 22
 23import structlog
 24from structlog.types import Processor
 25
 26from tiphys.config import LogLevel
 27
 28# Context variable for request-scoped logging
 29_log_context: ContextVar[dict[str, Any] | None] = ContextVar("log_context", default=None)
 30
 31
 32@dataclass
 33class LogContext:
 34    """
 35    Request-scoped logging context.
 36
 37    Provides structured context that propagates through the request lifecycle.
 38    All fields are automatically included in log output.
 39
 40    Example:
 41        from tiphys.logging import LogContext, set_log_context
 42
 43        # At request start
 44        ctx = LogContext(
 45            channel_name="webchat",
 46            peer_id="user123",
 47            session_key="webchat:user123",
 48        )
 49        set_log_context(ctx)
 50
 51        # All subsequent logs will include the context
 52        logger.info("Processing message")  # Includes correlation_id, channel, etc.
 53    """
 54
 55    correlation_id: str = field(default_factory=lambda: str(uuid4()))
 56    channel_name: str | None = None
 57    peer_id: str | None = None
 58    session_key: str | None = None
 59    agent_id: str | None = None
 60    message_id: str | None = None
 61    extra: dict[str, Any] = field(default_factory=dict)
 62
 63    def as_dict(self) -> dict[str, Any]:
 64        """Convert to dict for structlog binding."""
 65        result: dict[str, Any] = {"correlation_id": self.correlation_id}
 66        if self.channel_name:
 67            result["channel"] = self.channel_name
 68        if self.peer_id:
 69            result["peer_id"] = self.peer_id
 70        if self.session_key:
 71            result["session_key"] = self.session_key
 72        if self.agent_id:
 73            result["agent_id"] = self.agent_id
 74        if self.message_id:
 75            result["message_id"] = self.message_id
 76        result.update(self.extra)
 77        return result
 78
 79    def with_extra(self, **kwargs: Any) -> LogContext:
 80        """Return a new context with additional fields."""
 81        new_extra = {**self.extra, **kwargs}
 82        return LogContext(
 83            correlation_id=self.correlation_id,
 84            channel_name=self.channel_name,
 85            peer_id=self.peer_id,
 86            session_key=self.session_key,
 87            agent_id=self.agent_id,
 88            message_id=self.message_id,
 89            extra=new_extra,
 90        )
 91
 92
 93def set_log_context(ctx: LogContext) -> None:
 94    """
 95    Set the current logging context.
 96
 97    Args:
 98        ctx: The LogContext to set as current.
 99    """
100    _log_context.set(ctx.as_dict())
101
102
103def get_log_context() -> dict[str, Any]:
104    """
105    Get the current logging context.
106
107    Returns:
108        Dict of context values, or empty dict if not set.
109    """
110    return _log_context.get() or {}
111
112
113def add_log_context(**kwargs: Any) -> None:
114    """
115    Add fields to the current logging context.
116
117    Args:
118        **kwargs: Fields to add to the context.
119    """
120    ctx = (_log_context.get() or {}).copy()
121    ctx.update(kwargs)
122    _log_context.set(ctx)
123
124
125def clear_log_context() -> None:
126    """Clear the current logging context."""
127    _log_context.set({})
128
129
130def inject_context(
131    logger: Any,
132    method_name: str,
133    event_dict: MutableMapping[str, Any],
134) -> MutableMapping[str, Any]:
135    """
136    Structlog processor that injects request-scoped context.
137
138    Context values are injected but can be overridden by explicit log arguments.
139
140    Args:
141        logger: The bound logger.
142        method_name: The logging method name.
143        event_dict: The event dictionary.
144
145    Returns:
146        Event dict with context values merged.
147    """
148    ctx = get_log_context()
149    # Context values are defaults - explicit arguments override them
150    return {**ctx, **event_dict}
151
152
153def setup_logging(
154    level: LogLevel = LogLevel.INFO,
155    json_output: bool = False,
156) -> None:
157    """
158    Configure structured logging for Tiphys.
159
160    Args:
161        level: Minimum log level to output.
162        json_output: If True, output logs as JSON. Otherwise, use colored console output.
163    """
164    # Convert LogLevel enum to logging level
165    log_level = getattr(logging, level.value)
166
167    # Shared processors for all configurations
168    shared_processors: list[Processor] = [
169        structlog.contextvars.merge_contextvars,
170        inject_context,  # Inject request-scoped context (correlation_id, channel, etc.)
171        structlog.processors.add_log_level,
172        structlog.processors.TimeStamper(fmt="iso"),
173        structlog.stdlib.PositionalArgumentsFormatter(),
174        structlog.processors.StackInfoRenderer(),
175        structlog.processors.UnicodeDecoder(),
176    ]
177
178    if json_output:
179        # JSON output for production
180        processors: list[Processor] = [
181            *shared_processors,
182            structlog.processors.format_exc_info,
183            structlog.processors.JSONRenderer(),
184        ]
185    else:
186        # Colored console output for development
187        processors = [
188            *shared_processors,
189            structlog.dev.ConsoleRenderer(colors=True),
190        ]
191
192    # Configure structlog
193    structlog.configure(
194        processors=processors,
195        wrapper_class=structlog.make_filtering_bound_logger(log_level),
196        context_class=dict,
197        logger_factory=structlog.PrintLoggerFactory(),
198        cache_logger_on_first_use=True,
199    )
200
201    # Also configure stdlib logging for third-party libraries
202    logging.basicConfig(
203        format="%(message)s",
204        stream=sys.stdout,
205        level=log_level,
206    )
207
208    # Reduce noise from third-party libraries
209    logging.getLogger("uvicorn").setLevel(logging.WARNING)
210    logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
211    logging.getLogger("httpx").setLevel(logging.WARNING)
212    logging.getLogger("httpcore").setLevel(logging.WARNING)
213
214
215def get_logger(name: str) -> structlog.typing.FilteringBoundLogger:
216    """
217    Get a logger for a specific module.
218
219    Args:
220        name: Module name (typically __name__).
221
222    Returns:
223        A bound logger instance.
224    """
225    return structlog.get_logger(name)  # type: ignore[no-any-return]