Skip to content

Baggage

faststream.opentelemetry.baggage.Baggage #

Baggage(
    payload: dict[str, Any],
    batch_payload: list[dict[str, Any]] | None = None,
)
Source code in faststream/opentelemetry/baggage.py 
16
17
18
19
20
21
22
def __init__(
    self,
    payload: dict[str, Any],
    batch_payload: list[dict[str, Any]] | None = None,
) -> None:
    self._baggage = dict(payload)
    self._batch_baggage = [dict(b) for b in batch_payload] if batch_payload else []

get_all #

get_all() -> dict[str, Any]

Get a copy of the current baggage.

Source code in faststream/opentelemetry/baggage.py 
24
25
26
def get_all(self) -> dict[str, Any]:
    """Get a copy of the current baggage."""
    return self._baggage.copy()

get_all_batch #

get_all_batch() -> list[dict[str, Any]]

Get a copy of all batch baggage if exists.

Source code in faststream/opentelemetry/baggage.py 
28
29
30
def get_all_batch(self) -> list[dict[str, Any]]:
    """Get a copy of all batch baggage if exists."""
    return self._batch_baggage.copy()

get #

get(key: str) -> Any | None

Get a value from the baggage by key.

Source code in faststream/opentelemetry/baggage.py 
32
33
34
def get(self, key: str) -> Any | None:
    """Get a value from the baggage by key."""
    return self._baggage.get(key)

remove #

remove(key: str) -> None

Remove a baggage item by key.

Source code in faststream/opentelemetry/baggage.py 
36
37
38
def remove(self, key: str) -> None:
    """Remove a baggage item by key."""
    self._baggage.pop(key, None)

set #

set(key: str, value: Any) -> None

Set a key-value pair in the baggage.

Source code in faststream/opentelemetry/baggage.py 
40
41
42
def set(self, key: str, value: Any) -> None:
    """Set a key-value pair in the baggage."""
    self._baggage[key] = value

clear #

clear() -> None

Clear the current baggage.

Source code in faststream/opentelemetry/baggage.py 
44
45
46
def clear(self) -> None:
    """Clear the current baggage."""
    self._baggage.clear()

to_headers #

to_headers(
    headers: dict[str, Any] | None = None,
) -> dict[str, Any]

Convert baggage items to headers format for propagation.

Source code in faststream/opentelemetry/baggage.py 
48
49
50
51
52
53
54
55
56
57
58
def to_headers(self, headers: dict[str, Any] | None = None) -> dict[str, Any]:
    """Convert baggage items to headers format for propagation."""
    current_context = context.get_current()
    if headers is None:
        headers = {}

    for k, v in self._baggage.items():
        current_context = baggage.set_baggage(k, v, context=current_context)

    _BAGGAGE_PROPAGATOR.inject(headers, current_context)
    return headers

from_msg classmethod #

from_msg(msg: StreamMessage[Any]) -> Self

Create a Baggage instance from a StreamMessage.

Source code in faststream/opentelemetry/baggage.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@classmethod
def from_msg(cls, msg: "StreamMessage[Any]") -> Self:
    """Create a Baggage instance from a StreamMessage."""
    if len(msg.batch_headers) <= 1:
        return cls.from_headers(msg.headers)

    cumulative_baggage: dict[str, Any] = {}
    batch_baggage: list[dict[str, Any]] = []

    for headers in msg.batch_headers:
        payload = baggage.get_all(_BAGGAGE_PROPAGATOR.extract(headers))
        cumulative_baggage.update(payload)
        batch_baggage.append(cast("dict[str, Any]", payload))

    return cls(cumulative_baggage, batch_baggage)

from_headers classmethod #

from_headers(headers: dict[str, Any]) -> Self

Create a Baggage instance from headers.

Source code in faststream/opentelemetry/baggage.py 
76
77
78
79
80
@classmethod
def from_headers(cls, headers: dict[str, Any]) -> Self:
    """Create a Baggage instance from headers."""
    payload = baggage.get_all(_BAGGAGE_PROPAGATOR.extract(headers))
    return cls(cast("dict[str, Any]", payload))