Skip to content

Propagate

This module provides a thin wrapper around the OpenTelemetry propagate API to allow the OpenTelemetry contexts (and therefore Logfire contexts) to be transferred between different code running in different threads, processes or even services.

In general, you should not need to use this module since Logfire will automatically patch ThreadPoolExecutor and ProcessPoolExecutor to carry over the context. And existing plugins exist to propagate the context with requests and httpx.

get_context 

get_context() -> ContextCarrier

Create a new empty carrier dict and inject context into it.

Returns:

Type Description
ContextCarrier

A new dict with the context injected into it.

Usage:

from logfire.propagate import get_context, attach_context

logfire_context = get_context()

...

# later on in another thread, process or service
with attach_context(logfire_context):
    ...

You could also inject context into an existing mapping like headers with:

from logfire.propagate import get_context

existing_headers = {'X-Foobar': 'baz'}
existing_headers.update(get_context())
...
Source code in logfire/propagate.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def get_context() -> ContextCarrier:
    """Create a new empty carrier dict and inject context into it.

    Returns:
        A new dict with the context injected into it.

    Usage:

    ```py
    from logfire.propagate import get_context, attach_context

    logfire_context = get_context()

    ...

    # later on in another thread, process or service
    with attach_context(logfire_context):
        ...
    ```

    You could also inject context into an existing mapping like headers with:

    ```py
    from logfire.propagate import get_context

    existing_headers = {'X-Foobar': 'baz'}
    existing_headers.update(get_context())
    ...
    ```
    """
    carrier: ContextCarrier = {}
    propagate.inject(carrier)
    return carrier

attach_context 

attach_context(carrier: ContextCarrier, *, third_party: bool = False) -> Iterator[None]

Attach a context as generated by get_context to the current execution context.

Since attach_context is a context manager, it restores the previous context when exiting.

Set third_party to True if using this inside a library intended to be used by others. This will respect the distributed_tracing argument of logfire.configure(), so users will be warned about unintentional distributed tracing by default and they can suppress it. See Unintentional Distributed Tracing for more information.

Source code in logfire/propagate.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@contextmanager
def attach_context(carrier: ContextCarrier, *, third_party: bool = False) -> Iterator[None]:
    """Attach a context as generated by [`get_context`][logfire.propagate.get_context] to the current execution context.

    Since `attach_context` is a context manager, it restores the previous context when exiting.

    Set `third_party` to `True` if using this inside a library intended to be used by others.
    This will respect the [`distributed_tracing` argument of `logfire.configure()`][logfire.configure(distributed_tracing)],
    so users will be warned about unintentional distributed tracing by default and they can suppress it.
    See [Unintentional Distributed Tracing](https://logfire.pydantic.dev/docs/how-to-guides/distributed-tracing/#unintentional-distributed-tracing) for more information.
    """
    # capture the current context to restore it later
    old_context = otel_context.get_current()
    propagator = propagate.get_global_textmap()
    if not third_party:
        while isinstance(propagator, (WarnOnExtractTraceContextPropagator, NoExtractTraceContextPropagator)):
            propagator = propagator.wrapped
    new_context = propagator.extract(carrier=carrier)
    try:
        otel_context.attach(new_context)
        yield
    finally:
        otel_context.attach(old_context)