Skip to content

honeyhive.utils.baggage_dict

Baggage dictionary for OpenTelemetry context management.

BaggageDict

Dictionary-like interface for OpenTelemetry baggage.

This class provides a convenient way to work with OpenTelemetry baggage as if it were a regular dictionary, while maintaining proper context propagation.

Source code in src/honeyhive/utils/baggage_dict.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 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
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
class BaggageDict:
    """Dictionary-like interface for OpenTelemetry baggage.

    This class provides a convenient way to work with OpenTelemetry baggage
    as if it were a regular dictionary, while maintaining proper context
    propagation.
    """

    def __init__(self, ctx: Optional[Context] = None):
        """Initialize BaggageDict with optional context.

        Args:
            ctx: OpenTelemetry context. If None, uses current context.
        """

        self._context = ctx or context.get_current()

    @property
    def context(self) -> Context:
        """Get the current context."""
        return self._context

    def get(self, key: str, default: Any = None) -> Any:
        """Get a value from baggage.

        Args:
            key: Baggage key
            default: Default value if key not found

        Returns:
            Value from baggage or default
        """

        value = baggage.get_baggage(key, self._context)
        return value if value is not None else default

    def set(self, key: str, value: Any) -> "BaggageDict":
        """Set a value in baggage.

        Args:
            key: Baggage key
            value: Value to set

        Returns:
            New BaggageDict with updated context
        """

        new_context = baggage.set_baggage(key, str(value), self._context)
        return BaggageDict(new_context)

    def delete(self, key: str) -> "BaggageDict":
        """Delete a key from baggage.

        Args:
            key: Baggage key to delete

        Returns:
            New BaggageDict with updated context
        """

        new_context = baggage.set_baggage(key, None, self._context)
        return BaggageDict(new_context)

    def update(self, **kwargs: Any) -> "BaggageDict":
        """Update multiple baggage values.

        Args:
            **kwargs: Key-value pairs to set

        Returns:
            New BaggageDict with updated context
        """

        new_context = self._context
        for key, value in kwargs.items():
            new_context = baggage.set_baggage(key, str(value), new_context)

        return BaggageDict(new_context)

    def clear(self) -> "BaggageDict":
        """Clear all baggage.

        Returns:
            New BaggageDict with empty baggage
        """

        # Create new context without baggage
        new_context = context.get_current()
        return BaggageDict(new_context)

    def items(self) -> Dict[str, str]:
        """Get all baggage items as a dictionary.

        Returns:
            Dictionary of baggage key-value pairs
        """

        try:
            # Get current baggage context
            current_baggage = baggage.get_all()
            if current_baggage:
                # Convert to string values to match the expected return type
                return {k: str(v) for k, v in current_baggage.items()}
            return {}
        except Exception:
            return {}

    def keys(self) -> KeysView[str]:
        """Get all baggage keys."""
        return self.items().keys()

    def values(self) -> ValuesView[str]:
        """Get all baggage values."""
        return self.items().values()

    def __getitem__(self, key: str) -> str:
        """Get baggage value using bracket notation."""
        value = self.get(key)
        if value is None:
            raise KeyError(key)
        return str(value)  # Ensure we return a string

    def __setitem__(self, key: str, value: Any) -> None:
        """Set baggage value using bracket notation."""
        self.set(key, value)

    def __delitem__(self, key: str) -> None:
        """Delete baggage key using bracket notation."""
        self.delete(key)

    def __contains__(self, key: str) -> bool:
        """Check if key exists in baggage."""
        return self.get(key) is not None

    def __len__(self) -> int:
        """Get number of baggage items."""
        return len(self.items())

    def __iter__(self) -> Iterator[str]:
        """Iterate over baggage keys."""
        return iter(self.keys())

    def __repr__(self) -> str:
        """String representation."""
        items = self.items()
        return f"BaggageDict({items})"

    @classmethod
    def from_dict(
        cls, data: Dict[str, Any], ctx: Optional[Context] = None
    ) -> "BaggageDict":
        """Create BaggageDict from dictionary.

        Args:
            data: Dictionary of key-value pairs
            ctx: Optional OpenTelemetry context

        Returns:
            New BaggageDict with baggage from dictionary
        """
        baggage_dict = cls(ctx)
        return baggage_dict.update(**data)

    @contextmanager
    def as_context(self) -> Iterator[None]:
        """Context manager to temporarily set baggage in current context.

        Example:
            with BaggageDict().set("user_id", "123").as_context():
                # baggage is available in this context
                pass
        """
        token = context.attach(self._context)
        try:
            yield
        finally:
            context.detach(token)

context property 

context: Context

Get the current context.

get 

get(key: str, default: Any = None) -> Any

Get a value from baggage.

Parameters:

Name Type Description Default
key str

Baggage key

 required
default Any

Default value if key not found

 None

Returns:

Type Description
Any

Value from baggage or default
Source code in src/honeyhive/utils/baggage_dict.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
def get(self, key: str, default: Any = None) -> Any:
    """Get a value from baggage.

    Args:
        key: Baggage key
        default: Default value if key not found

    Returns:
        Value from baggage or default
    """

    value = baggage.get_baggage(key, self._context)
    return value if value is not None else default

set 

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

Set a value in baggage.

Parameters:

Name Type Description Default
key str

Baggage key

 required
value Any

Value to set

 required

Returns:

Type Description
BaggageDict

New BaggageDict with updated context
Source code in src/honeyhive/utils/baggage_dict.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
def set(self, key: str, value: Any) -> "BaggageDict":
    """Set a value in baggage.

    Args:
        key: Baggage key
        value: Value to set

    Returns:
        New BaggageDict with updated context
    """

    new_context = baggage.set_baggage(key, str(value), self._context)
    return BaggageDict(new_context)

delete 

delete(key: str) -> BaggageDict

Delete a key from baggage.

Parameters:

Name Type Description Default
key str

Baggage key to delete

 required

Returns:

Type Description
BaggageDict

New BaggageDict with updated context
Source code in src/honeyhive/utils/baggage_dict.py 
60
61
62
63
64
65
66
67
68
69
70
71
def delete(self, key: str) -> "BaggageDict":
    """Delete a key from baggage.

    Args:
        key: Baggage key to delete

    Returns:
        New BaggageDict with updated context
    """

    new_context = baggage.set_baggage(key, None, self._context)
    return BaggageDict(new_context)

update 

update(**kwargs: Any) -> BaggageDict

Update multiple baggage values.

Parameters:

Name Type Description Default
**kwargs Any

Key-value pairs to set

 {}

Returns:

Type Description
BaggageDict

New BaggageDict with updated context
Source code in src/honeyhive/utils/baggage_dict.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def update(self, **kwargs: Any) -> "BaggageDict":
    """Update multiple baggage values.

    Args:
        **kwargs: Key-value pairs to set

    Returns:
        New BaggageDict with updated context
    """

    new_context = self._context
    for key, value in kwargs.items():
        new_context = baggage.set_baggage(key, str(value), new_context)

    return BaggageDict(new_context)

clear 

clear() -> BaggageDict

Clear all baggage.

Returns:

Type Description
BaggageDict

New BaggageDict with empty baggage
Source code in src/honeyhive/utils/baggage_dict.py 
89
90
91
92
93
94
95
96
97
98
def clear(self) -> "BaggageDict":
    """Clear all baggage.

    Returns:
        New BaggageDict with empty baggage
    """

    # Create new context without baggage
    new_context = context.get_current()
    return BaggageDict(new_context)

items 

items() -> Dict[str, str]

Get all baggage items as a dictionary.

Returns:

Type Description
Dict[str, str]

Dictionary of baggage key-value pairs
Source code in src/honeyhive/utils/baggage_dict.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def items(self) -> Dict[str, str]:
    """Get all baggage items as a dictionary.

    Returns:
        Dictionary of baggage key-value pairs
    """

    try:
        # Get current baggage context
        current_baggage = baggage.get_all()
        if current_baggage:
            # Convert to string values to match the expected return type
            return {k: str(v) for k, v in current_baggage.items()}
        return {}
    except Exception:
        return {}

keys 

keys() -> KeysView[str]

Get all baggage keys.

Source code in src/honeyhive/utils/baggage_dict.py 
117
118
119
def keys(self) -> KeysView[str]:
    """Get all baggage keys."""
    return self.items().keys()

values 

values() -> ValuesView[str]

Get all baggage values.

Source code in src/honeyhive/utils/baggage_dict.py 
121
122
123
def values(self) -> ValuesView[str]:
    """Get all baggage values."""
    return self.items().values()

from_dict classmethod 

from_dict(
    data: Dict[str, Any], ctx: Optional[Context] = None
) -> BaggageDict

Create BaggageDict from dictionary.

Parameters:

Name Type Description Default
data Dict[str, Any]

Dictionary of key-value pairs

 required
ctx Optional[Context]

Optional OpenTelemetry context

 None

Returns:

Type Description
BaggageDict

New BaggageDict with baggage from dictionary
Source code in src/honeyhive/utils/baggage_dict.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
@classmethod
def from_dict(
    cls, data: Dict[str, Any], ctx: Optional[Context] = None
) -> "BaggageDict":
    """Create BaggageDict from dictionary.

    Args:
        data: Dictionary of key-value pairs
        ctx: Optional OpenTelemetry context

    Returns:
        New BaggageDict with baggage from dictionary
    """
    baggage_dict = cls(ctx)
    return baggage_dict.update(**data)

as_context 

as_context() -> Iterator[None]

Context manager to temporarily set baggage in current context.

Example

with BaggageDict().set("user_id", "123").as_context(): # baggage is available in this context pass

Source code in src/honeyhive/utils/baggage_dict.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@contextmanager
def as_context(self) -> Iterator[None]:
    """Context manager to temporarily set baggage in current context.

    Example:
        with BaggageDict().set("user_id", "123").as_context():
            # baggage is available in this context
            pass
    """
    token = context.attach(self._context)
    try:
        yield
    finally:
        context.detach(token)