Skip to content

donna.logging.setup

donna.logging.setup

Structured logging setup for Donna.

Configures structlog with JSON output and contextvars for async context propagation. All services import and call setup_logging() at startup. See docs/observability.md.

correlation_id_var module-attribute 

correlation_id_var: ContextVar[str] = ContextVar('correlation_id', default='')

user_id_var module-attribute 

user_id_var: ContextVar[str] = ContextVar('user_id', default='system')

channel_var module-attribute 

channel_var: ContextVar[str] = ContextVar('channel', default='')

task_id_var module-attribute 

task_id_var: ContextVar[str] = ContextVar('task_id', default='')

add_context_vars 

add_context_vars(logger: WrappedLogger, method_name: str, event_dict: EventDict) -> structlog.types.EventDict

Inject context variables into every log entry.

Source code in src/donna/logging/setup.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def add_context_vars(
    logger: structlog.types.WrappedLogger,
    method_name: str,
    event_dict: structlog.types.EventDict,
) -> structlog.types.EventDict:
    """Inject context variables into every log entry."""
    ctx_vars = {
        "correlation_id": correlation_id_var.get(""),
        "user_id": user_id_var.get("system"),
        "channel": channel_var.get(""),
        "task_id": task_id_var.get(""),
    }
    # Only include non-empty values
    for key, value in ctx_vars.items():
        if value:
            event_dict[key] = value
    return event_dict

setup_logging 

setup_logging(log_level: str = 'INFO', json_output: bool = True) -> None

Configure structured logging for all Donna services.

Parameters:

Name Type Description Default
log_level str

Minimum log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)

 'INFO'
json_output bool

If True, output JSON. If False, output human-readable (dev mode).

 True
Source code in src/donna/logging/setup.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def setup_logging(
    log_level: str = "INFO",
    json_output: bool = True,
) -> None:
    """Configure structured logging for all Donna services.

    Args:
        log_level: Minimum log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
        json_output: If True, output JSON. If False, output human-readable (dev mode).
    """
    shared_processors: list[structlog.types.Processor] = [
        structlog.contextvars.merge_contextvars,
        add_context_vars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
    ]

    renderer: structlog.types.Processor
    if json_output:
        renderer = structlog.processors.JSONRenderer()
    else:
        renderer = structlog.dev.ConsoleRenderer()

    structlog.configure(
        processors=[
            *shared_processors,
            structlog.processors.format_exc_info,
            renderer,
        ],
        wrapper_class=structlog.make_filtering_bound_logger(
            getattr(logging, log_level.upper(), logging.INFO)
        ),
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory(file=sys.stdout),
        cache_logger_on_first_use=True,
    )