12
12
# See the License for the specific language governing permissions and
13
13
# limitations under the License.
14
14
15
+ from __future__ import annotations
16
+
15
17
import urllib .parse
16
18
from contextlib import contextmanager
17
19
from importlib import import_module
18
20
from re import escape , sub
19
- from typing import Dict , Iterable , Sequence , Union
21
+ from typing import Any , Dict , Generator , Sequence
20
22
21
23
from wrapt import ObjectProxy
22
24
44
46
45
47
46
48
def extract_attributes_from_object (
47
- obj : any , attributes : Sequence [str ], existing : Dict [str , str ] = None
49
+ obj : Any , attributes : Sequence [str ], existing : Dict [str , str ] | None = None
48
50
) -> Dict [str , str ]:
49
- extracted = {}
51
+ extracted : dict [ str , str ] = {}
50
52
if existing :
51
53
extracted .update (existing )
52
54
for attr in attributes :
@@ -81,7 +83,7 @@ def http_status_to_status_code(
81
83
return StatusCode .ERROR
82
84
83
85
84
- def unwrap (obj : Union [ object , str ] , attr : str ):
86
+ def unwrap (obj : object , attr : str ):
85
87
"""Given a function that was wrapped by wrapt.wrap_function_wrapper, unwrap it
86
88
87
89
The object containing the function to unwrap may be passed as dotted module path string.
@@ -152,7 +154,7 @@ def _start_internal_or_server_span(
152
154
return span , token
153
155
154
156
155
- def _url_quote (s ) -> str : # pylint: disable=invalid-name
157
+ def _url_quote (s : Any ) -> str : # pylint: disable=invalid-name
156
158
if not isinstance (s , (str , bytes )):
157
159
return s
158
160
quoted = urllib .parse .quote (s )
@@ -163,13 +165,13 @@ def _url_quote(s) -> str: # pylint: disable=invalid-name
163
165
return quoted .replace ("%" , "%%" )
164
166
165
167
166
- def _get_opentelemetry_values () -> dict :
168
+ def _get_opentelemetry_values () -> dict [ str , Any ] :
167
169
"""
168
170
Return the OpenTelemetry Trace and Span IDs if Span ID is set in the
169
171
OpenTelemetry execution context.
170
172
"""
171
173
# Insert the W3C TraceContext generated
172
- _headers = {}
174
+ _headers : dict [ str , Any ] = {}
173
175
propagator .inject (_headers )
174
176
return _headers
175
177
@@ -196,7 +198,7 @@ def is_http_instrumentation_enabled() -> bool:
196
198
197
199
198
200
@contextmanager
199
- def _suppress_instrumentation (* keys : str ) -> Iterable [None ]:
201
+ def _suppress_instrumentation (* keys : str ) -> Generator [None ]:
200
202
"""Suppress instrumentation within the context."""
201
203
ctx = context .get_current ()
202
204
for key in keys :
@@ -209,7 +211,7 @@ def _suppress_instrumentation(*keys: str) -> Iterable[None]:
209
211
210
212
211
213
@contextmanager
212
- def suppress_instrumentation () -> Iterable [None ]:
214
+ def suppress_instrumentation () -> Generator [None ]:
213
215
"""Suppress instrumentation within the context."""
214
216
with _suppress_instrumentation (
215
217
_SUPPRESS_INSTRUMENTATION_KEY , _SUPPRESS_INSTRUMENTATION_KEY_PLAIN
@@ -218,7 +220,7 @@ def suppress_instrumentation() -> Iterable[None]:
218
220
219
221
220
222
@contextmanager
221
- def suppress_http_instrumentation () -> Iterable [None ]:
223
+ def suppress_http_instrumentation () -> Generator [None ]:
222
224
"""Suppress instrumentation within the context."""
223
225
with _suppress_instrumentation (_SUPPRESS_HTTP_INSTRUMENTATION_KEY ):
224
226
yield
0 commit comments