Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/opentelemetry/context/contextvars_context.py: 73%

1# Copyright The OpenTelemetry Authors 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14 

15from __future__ import annotations 

16 

17from contextvars import ContextVar, Token 

18 

19from opentelemetry.context.context import Context, _RuntimeContext 

20 

21 

22class ContextVarsRuntimeContext(_RuntimeContext): 

23 """An implementation of the RuntimeContext interface which wraps ContextVar under 

24 the hood. This is the preferred implementation for usage with Python 3.5+ 

25 """ 

26 

27 _CONTEXT_KEY = "current_context" 

28 

29 def __init__(self) -> None: 

30 self._current_context = ContextVar( 

31 self._CONTEXT_KEY, default=Context() 

32 ) 

33 

34 def attach(self, context: Context) -> Token[Context]: 

35 """Sets the current `Context` object. Returns a 

36 token that can be used to reset to the previous `Context`. 

37 

38 Args: 

39 context: The Context to set. 

40 """ 

41 return self._current_context.set(context) 

42 

43 def get_current(self) -> Context: 

44 """Returns the current `Context` object.""" 

45 return self._current_context.get() 

46 

47 def detach(self, token: Token[Context]) -> None: 

48 """Resets Context to a previous value 

49 

50 Args: 

51 token: A reference to a previous Context. 

52 """ 

53 self._current_context.reset(token) 

54 

55 

56__all__ = ["ContextVarsRuntimeContext"]