1# Copyright The OpenTelemetry Authors
2# SPDX-License-Identifier: Apache-2.0
3
4from __future__ import annotations
5
6from contextvars import ContextVar, Token
7
8from opentelemetry.context.context import Context, _RuntimeContext
9
10
11class ContextVarsRuntimeContext(_RuntimeContext):
12 """An implementation of the RuntimeContext interface which wraps ContextVar under
13 the hood. This is the preferred implementation for usage with Python 3.5+
14 """
15
16 _CONTEXT_KEY = "current_context"
17
18 def __init__(self) -> None:
19 self._current_context = ContextVar(
20 self._CONTEXT_KEY, default=Context()
21 )
22
23 def attach(self, context: Context) -> Token[Context]:
24 """Sets the current `Context` object. Returns a
25 token that can be used to reset to the previous `Context`.
26
27 Args:
28 context: The Context to set.
29 """
30 return self._current_context.set(context)
31
32 def get_current(self) -> Context:
33 """Returns the current `Context` object."""
34 return self._current_context.get()
35
36 def detach(self, token: Token[Context]) -> None:
37 """Resets Context to a previous value
38
39 Args:
40 token: A reference to a previous Context.
41 """
42 self._current_context.reset(token)
43
44
45__all__ = ["ContextVarsRuntimeContext"]