Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Copyright The OpenTelemetry Authors
2# SPDX-License-Identifier: Apache-2.0
4from __future__ import annotations
6import logging
7from contextvars import Token
8from os import environ
9from uuid import uuid4
11# pylint: disable=wrong-import-position
12from opentelemetry.context.context import Context, _RuntimeContext # noqa
13from opentelemetry.environment_variables import OTEL_PYTHON_CONTEXT
14from opentelemetry.util._importlib_metadata import entry_points
16logger = logging.getLogger(__name__)
19def _load_runtime_context() -> _RuntimeContext:
20 """Initialize the RuntimeContext
22 Returns:
23 An instance of RuntimeContext.
24 """
26 # FIXME use a better implementation of a configuration manager
27 # to avoid having to get configuration values straight from
28 # environment variables
29 default_context = "contextvars_context"
31 configured_context = environ.get(OTEL_PYTHON_CONTEXT, default_context) # type: str
33 try:
34 return next( # type: ignore
35 iter( # type: ignore
36 entry_points( # type: ignore
37 group="opentelemetry_context",
38 name=configured_context,
39 )
40 )
41 ).load()()
42 except Exception: # pylint: disable=broad-exception-caught
43 logger.exception(
44 "Failed to load context: %s, fallback to %s",
45 configured_context,
46 default_context,
47 )
48 return next( # type: ignore
49 iter( # type: ignore
50 entry_points( # type: ignore
51 group="opentelemetry_context",
52 name=default_context,
53 )
54 )
55 ).load()()
58_RUNTIME_CONTEXT = _load_runtime_context()
61def create_key(keyname: str) -> str:
62 """To allow cross-cutting concern to control access to their local state,
63 the RuntimeContext API provides a function which takes a keyname as input,
64 and returns a unique key.
65 Args:
66 keyname: The key name is for debugging purposes and is not required to be unique.
67 Returns:
68 A unique string representing the newly created key.
69 """
70 return keyname + "-" + str(uuid4())
73def get_value(key: str, context: Context | None = None) -> object:
74 """To access the local state of a concern, the RuntimeContext API
75 provides a function which takes a context and a key as input,
76 and returns a value.
78 Args:
79 key: The key of the value to retrieve.
80 context: The context from which to retrieve the value, if None, the current context is used.
82 Returns:
83 The value associated with the key.
84 """
85 return context.get(key) if context is not None else get_current().get(key)
88def set_value(
89 key: str, value: object, context: Context | None = None
90) -> Context:
91 """To record the local state of a cross-cutting concern, the
92 RuntimeContext API provides a function which takes a context, a
93 key, and a value as input, and returns an updated context
94 which contains the new value.
96 Args:
97 key: The key of the entry to set.
98 value: The value of the entry to set.
99 context: The context to copy, if None, the current context is used.
101 Returns:
102 A new `Context` containing the value set.
103 """
104 if context is None:
105 context = get_current()
106 new_values = context.copy()
107 new_values[key] = value
108 return Context(new_values)
111def get_current() -> Context:
112 """To access the context associated with program execution,
113 the Context API provides a function which takes no arguments
114 and returns a Context.
116 Returns:
117 The current `Context` object.
118 """
119 return _RUNTIME_CONTEXT.get_current()
122def attach(context: Context) -> Token[Context]:
123 """Associates a Context with the caller's current execution unit. Returns
124 a token that can be used to restore the previous Context.
126 Args:
127 context: The Context to set as current.
129 Returns:
130 A token that can be used with `detach` to reset the context.
131 """
132 return _RUNTIME_CONTEXT.attach(context)
135def detach(token: Token[Context]) -> None:
136 """Resets the Context associated with the caller's current execution unit
137 to the value it had before attaching a specified Context.
139 Args:
140 token: The Token that was returned by a previous call to attach a Context.
141 """
142 try:
143 _RUNTIME_CONTEXT.detach(token)
144 except Exception: # pylint: disable=broad-exception-caught
145 logger.exception("Failed to detach context")
148# FIXME This is a temporary location for the suppress instrumentation key.
149# Once the decision around how to suppress instrumentation is made in the
150# spec, this key should be moved accordingly.
151_ON_EMIT_RECURSION_COUNT_KEY = create_key("on_emit_recursion_count")
152_SUPPRESS_INSTRUMENTATION_KEY = create_key("suppress_instrumentation")
153_SUPPRESS_HTTP_INSTRUMENTATION_KEY = create_key(
154 "suppress_http_instrumentation"
155)
157__all__ = [
158 "Context",
159 "attach",
160 "create_key",
161 "detach",
162 "get_current",
163 "get_value",
164 "set_value",
165]