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
7import os
8from contextvars import Token
9from uuid import uuid4
11# pylint: disable=wrong-import-position
12from opentelemetry.context.context import Context, _RuntimeContext
13from opentelemetry.context.contextvars_context import ContextVarsRuntimeContext
14from opentelemetry.environment_variables import OTEL_PYTHON_CONTEXT
16logger = logging.getLogger(__name__)
19def _load_runtime_context() -> _RuntimeContext:
20 """Initialize the RuntimeContext
22 Returns:
23 An instance of RuntimeContext.
24 """
25 configured_context = os.environ.get(OTEL_PYTHON_CONTEXT)
26 if not configured_context:
27 return ContextVarsRuntimeContext()
29 # pylint: disable=import-outside-toplevel,no-name-in-module
30 from opentelemetry.util._importlib_metadata import ( # noqa: PLC0415
31 entry_points,
32 )
34 try:
35 return next(iter(entry_points(group="opentelemetry_context", name=configured_context))).load()()
36 except Exception: # pylint: disable=broad-exception-caught
37 logger.exception(
38 "Failed to load context: %s, falling back to contextvars_context",
39 configured_context,
40 )
41 return ContextVarsRuntimeContext()
44_RUNTIME_CONTEXT = _load_runtime_context()
47def create_key(keyname: str) -> str:
48 """To allow cross-cutting concern to control access to their local state,
49 the RuntimeContext API provides a function which takes a keyname as input,
50 and returns a unique key.
51 Args:
52 keyname: The key name is for debugging purposes and is not required to be unique.
53 Returns:
54 A unique string representing the newly created key.
55 """
56 return keyname + "-" + str(uuid4())
59def get_value(key: str, context: Context | None = None) -> object:
60 """To access the local state of a concern, the RuntimeContext API
61 provides a function which takes a context and a key as input,
62 and returns a value.
64 Args:
65 key: The key of the value to retrieve.
66 context: The context from which to retrieve the value, if None, the current context is used.
68 Returns:
69 The value associated with the key.
70 """
71 return context.get(key) if context is not None else get_current().get(key)
74def set_value(key: str, value: object, context: Context | None = None) -> Context:
75 """To record the local state of a cross-cutting concern, the
76 RuntimeContext API provides a function which takes a context, a
77 key, and a value as input, and returns an updated context
78 which contains the new value.
80 Args:
81 key: The key of the entry to set.
82 value: The value of the entry to set.
83 context: The context to copy, if None, the current context is used.
85 Returns:
86 A new `Context` containing the value set.
87 """
88 if context is None:
89 context = get_current()
90 new_values = context.copy()
91 new_values[key] = value
92 return Context(new_values)
95def get_current() -> Context:
96 """To access the context associated with program execution,
97 the Context API provides a function which takes no arguments
98 and returns a Context.
100 Returns:
101 The current `Context` object.
102 """
103 return _RUNTIME_CONTEXT.get_current()
106def attach(context: Context) -> Token[Context]:
107 """Associates a Context with the caller's current execution unit. Returns
108 a token that can be used to restore the previous Context.
110 Args:
111 context: The Context to set as current.
113 Returns:
114 A token that can be used with `detach` to reset the context.
115 """
116 return _RUNTIME_CONTEXT.attach(context)
119def detach(token: Token[Context]) -> None:
120 """Resets the Context associated with the caller's current execution unit
121 to the value it had before attaching a specified Context.
123 Args:
124 token: The Token that was returned by a previous call to attach a Context.
125 """
126 try:
127 _RUNTIME_CONTEXT.detach(token)
128 except Exception: # pylint: disable=broad-exception-caught
129 logger.exception("Failed to detach context")
132# FIXME This is a temporary location for the suppress instrumentation key.
133# Once the decision around how to suppress instrumentation is made in the
134# spec, this key should be moved accordingly.
135_ON_EMIT_RECURSION_COUNT_KEY = create_key("on_emit_recursion_count")
136_SUPPRESS_INSTRUMENTATION_KEY = create_key("suppress_instrumentation")
137_SUPPRESS_HTTP_INSTRUMENTATION_KEY = create_key("suppress_http_instrumentation")
139__all__ = [
140 "Context",
141 "attach",
142 "create_key",
143 "detach",
144 "get_current",
145 "get_value",
146 "set_value",
147]