Last active
March 14, 2026 17:54
-
-
Save szapp/2caa8c23a64e909c3987226ed6a79cab to your computer and use it in GitHub Desktop.
Simple context in Python's stdlib logging
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env -S uv run --script | |
| # /// script | |
| # requires-python = ">=3.14" | |
| # dependencies = [ | |
| # "python-json-logger>=4.0.0", | |
| # ] | |
| # /// | |
| import logging | |
| import logging.config | |
| import warnings | |
| from collections.abc import Generator | |
| from contextlib import contextmanager | |
| from contextvars import ContextVar | |
| from typing import Any | |
| _log_context = ContextVar("log_context", default={}) | |
| def bind_logvars(**values: Any) -> dict[str, Any]: | |
| """Bind one or more logging variables to be attached to all logs going forward.""" | |
| ctx = _log_context.get().copy() | |
| values_backup = {k: ctx[k] for k in ctx.keys() & values.keys()} | |
| ctx.update(values) | |
| _log_context.set(ctx) | |
| return values_backup | |
| def unbind_logvars(*keys: str) -> None: | |
| """Remove one or multiple logging variables by their names.""" | |
| ctx = _log_context.get().copy() | |
| for k in keys: | |
| ctx.pop(k, None) | |
| _log_context.set(ctx) | |
| def clear_logvars(): | |
| """Clear all bound logging variables for testing.""" | |
| _log_context.set({}) | |
| @contextmanager | |
| def bound_logvars(**values: Any) -> Generator[None, None, None]: | |
| """Temporarily bind and unbind/restore logging variables using with or decorator.""" | |
| backup = bind_logvars(**values) | |
| try: | |
| yield | |
| finally: | |
| unbind_logvars(*values) | |
| bind_logvars(**backup) | |
| class LoggingVarsFilter(logging.Filter): | |
| """Inject logging variables into log record.""" | |
| def filter(self, record: logging.LogRecord) -> bool: | |
| record.__dict__.update(_log_context.get()) | |
| return True | |
| class LoggingVarsFormatter(logging.Formatter): | |
| """Inject logging variables into log record. For demonstration only.""" | |
| def format(self, record: logging.LogRecord) -> str: | |
| record.context = ", ".join(f"{k}={v}" for k, v in _log_context.get().items()) | |
| return super().format(record) | |
| """ | |
| Config | |
| """ | |
| logging_config = { | |
| "version": 1, | |
| "disable_existing_loggers": False, | |
| "filters": { | |
| "context": { | |
| "()": "__main__.LoggingVarsFilter", | |
| }, | |
| }, | |
| "formatters": { | |
| "context": { | |
| "()": "__main__.LoggingVarsFormatter", | |
| "format": "[ %(levelname)s ] %(name)s %(message)s %(context)s", | |
| }, | |
| "structured": { | |
| "()": "pythonjsonlogger.json.JsonFormatter", | |
| "fmt": [ | |
| "timestamp", | |
| "levelname", | |
| "message", | |
| "name", | |
| "funcName", | |
| "lineno", | |
| ], | |
| "timestamp": True, | |
| "rename_fields": {"levelname": "status"}, | |
| "exc_info_as_array": False, | |
| "stack_info_as_array": False, | |
| }, | |
| }, | |
| "handlers": { | |
| "monitoring": { | |
| "()": "logging.handlers.RotatingFileHandler", | |
| "level": "INFO", | |
| "filters": ["context"], | |
| "formatter": "structured", | |
| "filename": "structured.log", | |
| "encoding": "utf-8", | |
| "maxBytes": 1024**2, # 1 MB | |
| "backupCount": 1, # Important for safe rotation (structured.log.1) | |
| }, | |
| "console": { | |
| "class": "logging.StreamHandler", | |
| "level": "INFO", | |
| "formatter": "context", | |
| "stream": "ext://sys.stdout", | |
| }, | |
| }, | |
| "root": { | |
| "level": "INFO", | |
| "handlers": ["monitoring", "console"], | |
| }, | |
| } | |
| logging.config.dictConfig(logging_config) | |
| logging.captureWarnings(True) | |
| """ | |
| Demo | |
| Output: | |
| [ INFO ] __main__ This message has a bound variable var=hi | |
| [ INFO ] __main__ This message does, too var=hi | |
| [ INFO ] __main__ Hello from myfunc! request_id=abc123, user=alice, task=myfunc | |
| [ INFO ] __main__ Temporary override request_id=abc123, user=2433 | |
| [ INFO ] __main__ Looping request_id=abc123, loop_iter=0 | |
| [ INFO ] __main__ Looping request_id=abc123, loop_iter=1 | |
| [ INFO ] __main__ Looping request_id=abc123, loop_iter=2 | |
| [ INFO ] __main__ With world request_id=abc123, hello=world | |
| [ INFO ] __main__ With space request_id=abc123, hello=space | |
| [ INFO ] __main__ World restored request_id=abc123, hello=world | |
| [ INFO ] __main__ Clean again | |
| """ | |
| logger = logging.getLogger(__name__) | |
| # Simple bind and unbind | |
| bind_logvars(var="hi") | |
| logger.info("This message has a bound variable") | |
| logger.info("This message does, too") | |
| unbind_logvars("var") | |
| # Normal message with extra variables without binding still work | |
| logger.info("Normal message", extra={"my_var": True}) | |
| # Attach a logging variable to the entire execution of a function | |
| @bound_logvars(task="myfunc") | |
| def myfunc(): | |
| logger.info("Hello from myfunc!") | |
| # Attach logging variables to the logs of all enclosed operations | |
| with bound_logvars(request_id="abc123", user="alice"): | |
| myfunc() | |
| # Overwrite existing logging variable and completely detach it | |
| bind_logvars(user=2433) | |
| logger.info("Temporary override") | |
| unbind_logvars("user") # User is now completely gone! | |
| # Monitor loop iterations with bound logging variable | |
| with bound_logvars(loop_iter=None): | |
| for i in range(3): | |
| bind_logvars(loop_iter=i) | |
| logger.info("Looping") | |
| # Bound context restores overridden variables | |
| with bound_logvars(hello="world"): | |
| logger.info("With world") | |
| with bound_logvars(hello="space"): | |
| logger.info("With space") | |
| logger.info("World restored") | |
| logger.info("Clean again") | |
| warnings.warn("hello") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment