Skip to content

Context Propagation

PyLogShield supports thread-safe and asyncio-safe structured log context injection via Python's contextvars module. Any key/value pairs set inside a log_context or async_log_context block are automatically attached to every log record emitted by a logger that has enable_context=True.

Quick Reference

Python
from pylogshield import get_logger
from pylogshield.context import log_context, async_log_context

logger = get_logger("app", enable_context=True, enable_json=True)

# Sync context
with log_context(request_id="abc-123", user_id=42):
    logger.info("Processing order")
    # JSON output includes request_id and user_id

# Async context
async with async_log_context(request_id="xyz-999"):
    logger.info("Async handler")

How It Works

When enable_context=True is passed to get_logger() or PyLogShield(), a ContextFilter is installed on the logger. The filter reads the current context dict from a ContextVar and injects each key directly onto the LogRecord before it reaches any formatter or handler.

With JsonFormatter, context fields are promoted to the top level of the JSON envelope alongside timestamp, level, and message.

sequenceDiagram
    participant App
    participant CM as log_context(...)
    participant CV as ContextVar
    participant CF as ContextFilter
    participant LR as LogRecord
    participant FMT as JsonFormatter

    App->>CM: enter with log_context(request_id="abc")
    CM->>CV: ContextVar.set({"request_id": "abc"})
    App->>CF: logger.info("Processing")
    CF->>CV: get_log_context()
    CV-->>CF: {"request_id": "abc"}
    CF->>LR: record.request_id = "abc"
    LR->>FMT: format(record)
    FMT-->>App: {"message": "Processing", "request_id": "abc", ...}
    App->>CM: exit context block
    CM->>CV: ContextVar.reset(token)

log_context

Sync context manager. Fields are merged on top of any already-active context, so nesting works as expected. The previous context is always restored on exit—even if an exception is raised.

Python
from pylogshield.context import log_context, get_log_context

logger = get_logger("api", enable_context=True, enable_json=True)

with log_context(service="payments"):
    logger.info("Payment service started")
    # → {"service": "payments", "message": "Payment service started", ...}

    with log_context(transaction_id="tx-7", amount=99.99):
        logger.info("Charge applied")
        # → {"service": "payments", "transaction_id": "tx-7", "amount": 99.99, ...}
        print(get_log_context())
        # {'service': 'payments', 'transaction_id': 'tx-7', 'amount': 99.99}

    logger.info("Transaction complete")
    # → {"service": "payments", "message": "Transaction complete", ...}
    # transaction_id is gone; service is restored

logger.info("Done")  # no context fields at all

Behaviour when an exception is raised inside the block:

Python
with log_context(service="billing"):
    try:
        with log_context(step="charge"):
            raise RuntimeError("card declined")
    except RuntimeError:
        logger.exception("Charge failed")
        # → {"service": "billing", "message": "Charge failed", ...}
        # 'step' is gone even though the inner block raised — the context is
        # always restored by the finally clause inside log_context.

async_log_context

Async context manager. Each asyncio task gets its own copy of the ContextVar, so concurrent tasks cannot bleed context into each other.

Python
import asyncio
from pylogshield.context import async_log_context

async def handle_request(request_id: str):
    async with async_log_context(request_id=request_id):
        logger.info("Handling request")  # carries request_id

# Safe with asyncio.gather — no cross-task bleed
await asyncio.gather(
    handle_request("req-1"),
    handle_request("req-2"),
)

ContextFilter

The filter that connects the ContextVar to the logging system. Added automatically when enable_context=True; can also be added manually:

Python
from pylogshield.context import ContextFilter, log_context
import logging

logger = logging.getLogger("my_app")
logger.addFilter(ContextFilter())

with log_context(env="prod"):
    logger.info("Deployed")  # record.env == "prod"

Reserved field names

Context keys that conflict with standard LogRecord attributes (e.g., name, msg, levelname, exc_info, args) are silently skipped and a warnings.warn is emitted once per conflicting key per ContextFilter instance. The warning message suggests a safe rename — for example, using name as a context key would warn:

Text Only
pylogshield: context key 'name' conflicts with a standard LogRecord attribute
and will be ignored. Use a different name (e.g. 'ctx_name') to avoid this conflict.

Rename your key (e.g. service_name, logger_name) to avoid this.

get_log_context

Returns the current context dict for the active thread or asyncio task. Returns an empty dict when no context block is active.

Python
from pylogshield.context import get_log_context

ctx = get_log_context()
print(ctx)  # {'request_id': 'abc-123', 'user_id': 42}

Production Example (with JSON)

Python
from pylogshield import get_logger
from pylogshield.context import log_context

logger = get_logger("api", enable_context=True, enable_json=True)

def process_payment(user_id: int, amount: float):
    with log_context(user_id=user_id, operation="payment"):
        logger.info("Payment initiated", extra={"amount": amount})
        # ...
        logger.info("Payment complete")

JSON output:

JSON
{"timestamp": "...", "level": "INFO", "message": "Payment initiated", "user_id": 123, "operation": "payment", "amount": 99.99}

API Reference

context

Context propagation for PyLogShield using Python's contextvars.

This module provides thread-safe and asyncio-safe log context injection via contextvars.ContextVar. Any key/value pairs set inside a log_context or async_log_context block are automatically attached to every log record emitted by a logger that has enable_context=True (or that has a :class:ContextFilter installed).

Usage::

Text Only
from pylogshield import get_logger
from pylogshield.context import log_context, async_log_context

logger = get_logger("app", enable_context=True, enable_json=True)

# Sync
with log_context(request_id="abc-123", user_id=42):
    logger.info("Processing order")  # JSON includes request_id + user_id

# Async
async with async_log_context(request_id="xyz-999"):
    logger.info("Async handler")

# Nested — inner fields are merged on top of outer fields
with log_context(service="payments"):
    with log_context(transaction_id="tx-7"):
        logger.info("Charge applied")  # has both fields
    logger.info("Done")               # only service remains
CLASS DESCRIPTION
ContextFilter

Logging filter that injects the active log context into every LogRecord.
FUNCTION DESCRIPTION
log_context

Sync context manager that injects fields into every log within the block.
async_log_context

Async context manager that injects fields into every log within the block.
get_log_context

Return the current log context dict for this thread / asyncio task.

ContextFilter(name: str = '')

Bases: Filter

Logging filter that injects the active log context into every LogRecord.

Add this filter to a :class:~pylogshield.PyLogShield instance (or any logging.Logger / logging.Handler) to have context variables automatically stamped onto log records.

When :class:~pylogshield.handlers.JsonFormatter is used the context fields are promoted to the top level of the JSON envelope (alongside timestamp, level, etc.) rather than being nested under extra.

Notes

This filter always returns True — it never suppresses records. Context keys that conflict with standard LogRecord attribute names are silently skipped (a warnings.warn is emitted once per conflicting key).

Examples:

Python Console Session
>>> from pylogshield.context import ContextFilter, log_context
>>> logger.addFilter(ContextFilter())
>>> with log_context(env="prod"):
...     logger.info("deployed")   # record.env == "prod"
Source code in src/pylogshield/context.py
Python
169
170
171
172
def __init__(self, name: str = "") -> None:
    super().__init__(name)
    self._warned_keys: set = set()
    self._warned_keys_lock = threading.Lock()

log_context(**fields: Any) -> Iterator[None]

Sync context manager that injects fields into every log within the block.

Fields are merged on top of any already-active context, so nesting works as expected. The previous context is restored on exit (including on exceptions).

PARAMETER DESCRIPTION

**fields

Arbitrary key/value pairs to attach to log records.

TYPE: Any DEFAULT: {}

Examples:

Python Console Session
>>> with log_context(request_id="abc", user_id=42):
...     logger.info("Processing")   # record carries request_id and user_id
Source code in src/pylogshield/context.py
Python
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@contextmanager
def log_context(**fields: Any) -> Iterator[None]:
    """Sync context manager that injects *fields* into every log within the block.

    Fields are merged on top of any already-active context, so nesting works
    as expected.  The previous context is restored on exit (including on
    exceptions).

    Parameters
    ----------
    **fields : Any
        Arbitrary key/value pairs to attach to log records.

    Examples
    --------
    >>> with log_context(request_id="abc", user_id=42):
    ...     logger.info("Processing")   # record carries request_id and user_id
    """
    token: Token[Optional[Dict[str, Any]]] = _log_context.set(
        {**(_log_context.get() or {}), **fields}
    )
    try:
        yield
    finally:
        _log_context.reset(token)

async_log_context(**fields: Any) -> AsyncIterator[None] async

Async context manager that injects fields into every log within the block.

Safe to use with asyncio.gather — each task gets its own copy of the ContextVar so contexts do not bleed between concurrent tasks.

PARAMETER DESCRIPTION

**fields

Arbitrary key/value pairs to attach to log records.

TYPE: Any DEFAULT: {}

Examples:

Python Console Session
>>> async with async_log_context(request_id="xyz"):
...     logger.info("Async handler")  # record carries request_id
Source code in src/pylogshield/context.py
Python
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@asynccontextmanager
async def async_log_context(**fields: Any) -> AsyncIterator[None]:
    """Async context manager that injects *fields* into every log within the block.

    Safe to use with ``asyncio.gather`` — each task gets its own copy of the
    ``ContextVar`` so contexts do not bleed between concurrent tasks.

    Parameters
    ----------
    **fields : Any
        Arbitrary key/value pairs to attach to log records.

    Examples
    --------
    >>> async with async_log_context(request_id="xyz"):
    ...     logger.info("Async handler")  # record carries request_id
    """
    token: Token[Optional[Dict[str, Any]]] = _log_context.set(
        {**(_log_context.get() or {}), **fields}
    )
    try:
        yield
    finally:
        _log_context.reset(token)

get_log_context() -> Dict[str, Any]

Return the current log context dict for this thread / asyncio task.

Returns an empty dict when no context block is active.

RETURNS DESCRIPTION
dict

The active context key/value pairs.
Source code in src/pylogshield/context.py
Python
50
51
52
53
54
55
56
57
58
59
60
def get_log_context() -> Dict[str, Any]:
    """Return the current log context dict for this thread / asyncio task.

    Returns an empty dict when no context block is active.

    Returns
    -------
    dict
        The active context key/value pairs.
    """
    return _log_context.get() or {}