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

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 

17import logging 

18import typing 

19from contextvars import Token 

20from os import environ 

21from uuid import uuid4 

22 

23# pylint: disable=wrong-import-position 

24from opentelemetry.context.context import Context, _RuntimeContext # noqa 

25from opentelemetry.environment_variables import OTEL_PYTHON_CONTEXT 

26from opentelemetry.util._importlib_metadata import entry_points 

27 

28logger = logging.getLogger(__name__) 

29 

30 

31def _load_runtime_context() -> _RuntimeContext: 

32 """Initialize the RuntimeContext 

33 

34 Returns: 

35 An instance of RuntimeContext. 

36 """ 

37 

38 # FIXME use a better implementation of a configuration manager 

39 # to avoid having to get configuration values straight from 

40 # environment variables 

41 default_context = "contextvars_context" 

42 

43 configured_context = environ.get(OTEL_PYTHON_CONTEXT, default_context) # type: str 

44 

45 try: 

46 return next( # type: ignore 

47 iter( # type: ignore 

48 entry_points( # type: ignore 

49 group="opentelemetry_context", 

50 name=configured_context, 

51 ) 

52 ) 

53 ).load()() 

54 except Exception: # pylint: disable=broad-exception-caught 

55 logger.exception( 

56 "Failed to load context: %s, fallback to %s", 

57 configured_context, 

58 default_context, 

59 ) 

60 return next( # type: ignore 

61 iter( # type: ignore 

62 entry_points( # type: ignore 

63 group="opentelemetry_context", 

64 name=default_context, 

65 ) 

66 ) 

67 ).load()() 

68 

69 

70_RUNTIME_CONTEXT = _load_runtime_context() 

71 

72 

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

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

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

76 and returns a unique key. 

77 Args: 

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

79 Returns: 

80 A unique string representing the newly created key. 

81 """ 

82 return keyname + "-" + str(uuid4()) 

83 

84 

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

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

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

88 and returns a value. 

89 

90 Args: 

91 key: The key of the value to retrieve. 

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

93 

94 Returns: 

95 The value associated with the key. 

96 """ 

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

98 

99 

100def set_value( 

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

102) -> Context: 

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

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

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

106 which contains the new value. 

107 

108 Args: 

109 key: The key of the entry to set. 

110 value: The value of the entry to set. 

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

112 

113 Returns: 

114 A new `Context` containing the value set. 

115 """ 

116 if context is None: 

117 context = get_current() 

118 new_values = context.copy() 

119 new_values[key] = value 

120 return Context(new_values) 

121 

122 

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() 

132 

133 

134def attach(context: Context) -> Token[Context]: 

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

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

137 

138 Args: 

139 context: The Context to set as current. 

140 

141 Returns: 

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

143 """ 

144 return _RUNTIME_CONTEXT.attach(context) 

145 

146 

147def detach(token: Token[Context]) -> None: 

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

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

150 

151 Args: 

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

153 """ 

154 try: 

155 _RUNTIME_CONTEXT.detach(token) 

156 except Exception: # pylint: disable=broad-exception-caught 

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

158 

159 

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

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

162# spec, this key should be moved accordingly. 

163_SUPPRESS_INSTRUMENTATION_KEY = create_key("suppress_instrumentation") 

164_SUPPRESS_HTTP_INSTRUMENTATION_KEY = create_key( 

165 "suppress_http_instrumentation" 

166) 

167 

168__all__ = [ 

169 "Context", 

170 "attach", 

171 "create_key", 

172 "detach", 

173 "get_current", 

174 "get_value", 

175 "set_value", 

176]