Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/opentelemetry/context/__init__.py: 58%

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 

15import logging 

16import threading 

17import typing 

18import uuid 

19from functools import wraps 

20from os import environ 

21 

22from pkg_resources import iter_entry_points 

23 

24from opentelemetry.context.context import Context, _RuntimeContext 

25from opentelemetry.environment_variables import OTEL_PYTHON_CONTEXT 

26 

27logger = logging.getLogger(__name__) 

28_RUNTIME_CONTEXT = None # type: typing.Optional[_RuntimeContext] 

29_RUNTIME_CONTEXT_LOCK = threading.Lock() 

30 

31_F = typing.TypeVar("_F", bound=typing.Callable[..., typing.Any]) 

32 

33 

34def _load_runtime_context(func: _F) -> _F: 

35 """A decorator used to initialize the global RuntimeContext 

36 

37 Returns: 

38 A wrapper of the decorated method. 

39 """ 

40 

41 @wraps(func) # type: ignore[misc] 

42 def wrapper( # type: ignore[misc] 

43 *args: typing.Tuple[typing.Any, typing.Any], 

44 **kwargs: typing.Dict[typing.Any, typing.Any] 

45 ) -> typing.Optional[typing.Any]: 

46 global _RUNTIME_CONTEXT # pylint: disable=global-statement 

47 

48 with _RUNTIME_CONTEXT_LOCK: 

49 if _RUNTIME_CONTEXT is None: 

50 # FIXME use a better implementation of a configuration manager to avoid having 

51 # to get configuration values straight from environment variables 

52 default_context = "contextvars_context" 

53 

54 configured_context = environ.get( 

55 OTEL_PYTHON_CONTEXT, default_context 

56 ) # type: str 

57 try: 

58 _RUNTIME_CONTEXT = next( 

59 iter_entry_points( 

60 "opentelemetry_context", configured_context 

61 ) 

62 ).load()() 

63 except Exception: # pylint: disable=broad-except 

64 logger.error( 

65 "Failed to load context: %s", configured_context 

66 ) 

67 return func(*args, **kwargs) # type: ignore[misc] 

68 

69 return typing.cast(_F, wrapper) # type: ignore[misc] 

70 

71 

72def create_key(keyname: str) -> str: 

73 """To allow cross-cutting concern to control access to their local state, 

74 the RuntimeContext API provides a function which takes a keyname as input, 

75 and returns a unique key. 

76 Args: 

77 keyname: The key name is for debugging purposes and is not required to be unique. 

78 Returns: 

79 A unique string representing the newly created key. 

80 """ 

81 return keyname + "-" + str(uuid.uuid4()) 

82 

83 

84def get_value(key: str, context: typing.Optional[Context] = None) -> "object": 

85 """To access the local state of a concern, the RuntimeContext API 

86 provides a function which takes a context and a key as input, 

87 and returns a value. 

88 

89 Args: 

90 key: The key of the value to retrieve. 

91 context: The context from which to retrieve the value, if None, the current context is used. 

92 

93 Returns: 

94 The value associated with the key. 

95 """ 

96 return context.get(key) if context is not None else get_current().get(key) 

97 

98 

99def set_value( 

100 key: str, value: "object", context: typing.Optional[Context] = None 

101) -> Context: 

102 """To record the local state of a cross-cutting concern, the 

103 RuntimeContext API provides a function which takes a context, a 

104 key, and a value as input, and returns an updated context 

105 which contains the new value. 

106 

107 Args: 

108 key: The key of the entry to set. 

109 value: The value of the entry to set. 

110 context: The context to copy, if None, the current context is used. 

111 

112 Returns: 

113 A new `Context` containing the value set. 

114 """ 

115 if context is None: 

116 context = get_current() 

117 new_values = context.copy() 

118 new_values[key] = value 

119 return Context(new_values) 

120 

121 

122@_load_runtime_context # type: ignore 

123def get_current() -> Context: 

124 """To access the context associated with program execution, 

125 the Context API provides a function which takes no arguments 

126 and returns a Context. 

127 

128 Returns: 

129 The current `Context` object. 

130 """ 

131 return _RUNTIME_CONTEXT.get_current() # type:ignore 

132 

133 

134@_load_runtime_context # type: ignore 

135def attach(context: Context) -> object: 

136 """Associates a Context with the caller's current execution unit. Returns 

137 a token that can be used to restore the previous Context. 

138 

139 Args: 

140 context: The Context to set as current. 

141 

142 Returns: 

143 A token that can be used with `detach` to reset the context. 

144 """ 

145 return _RUNTIME_CONTEXT.attach(context) # type:ignore 

146 

147 

148@_load_runtime_context # type: ignore 

149def detach(token: object) -> None: 

150 """Resets the Context associated with the caller's current execution unit 

151 to the value it had before attaching a specified Context. 

152 

153 Args: 

154 token: The Token that was returned by a previous call to attach a Context. 

155 """ 

156 try: 

157 _RUNTIME_CONTEXT.detach(token) # type: ignore 

158 except Exception: # pylint: disable=broad-except 

159 logger.exception("Failed to detach context") 

160 

161 

162# FIXME This is a temporary location for the suppress instrumentation key. 

163# Once the decision around how to suppress instrumentation is made in the 

164# spec, this key should be moved accordingly. 

165_SUPPRESS_INSTRUMENTATION_KEY = create_key("suppress_instrumentation") 

166_SUPPRESS_HTTP_INSTRUMENTATION_KEY = create_key( 

167 "suppress_http_instrumentation" 

168)