Baggage

faststream.opentelemetry.Baggage

Baggage(payload, batch_payload=None)

Source code in faststream/opentelemetry/baggage.py

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()

Get a copy of the current baggage.

Source code in faststream/opentelemetry/baggage.py

def get_all(self) -> dict[str, Any]:
    """Get a copy of the current baggage."""
    return self._baggage.copy()

get_all_batch

get_all_batch()

Get a copy of all batch baggage if exists.

Source code in faststream/opentelemetry/baggage.py

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)

Get a value from the baggage by key.

Source code in faststream/opentelemetry/baggage.py

def get(self, key: str) -> Any | None:
    """Get a value from the baggage by key."""
    return self._baggage.get(key)

remove

remove(key)

Remove a baggage item by key.

Source code in faststream/opentelemetry/baggage.py

def remove(self, key: str) -> None:
    """Remove a baggage item by key."""
    self._baggage.pop(key, None)

set

set(key, value)

Set a key-value pair in the baggage.

Source code in faststream/opentelemetry/baggage.py

def set(self, key: str, value: Any) -> None:
    """Set a key-value pair in the baggage."""
    self._baggage[key] = value

clear

clear()

Clear the current baggage.

Source code in faststream/opentelemetry/baggage.py

def clear(self) -> None:
    """Clear the current baggage."""
    self._baggage.clear()

to_headers

to_headers(headers=None)

Convert baggage items to headers format for propagation.

Source code in faststream/opentelemetry/baggage.py

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)

Create a Baggage instance from a StreamMessage.

Source code in faststream/opentelemetry/baggage.py

@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)

Create a Baggage instance from headers.

Source code in faststream/opentelemetry/baggage.py

@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))