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

1import asyncio 

2import contextvars 

3import threading 

4from typing import Any, final 

5 

6 

7class _Storage: 

8 """Thread-tagged storage for a non-thread-critical ``Local``. 

9 

10 The data is tagged with the identity of the thread that owns it. This lets 

11 ``_CVar`` ignore data that leaked into an unrelated thread. 

12 

13 Python 3.14 added ``sys.flags.thread_inherit_context``, which is enabled by 

14 default on free-threaded builds. When set, a new thread starts with a copy 

15 of the spawning thread's context instead of an empty one, so the contextvar 

16 backing a ``Local`` would otherwise be visible in any thread spawned from 

17 one that had set it -- breaking the documented "thread-local in sync 

18 threads" behaviour. asgiref re-homes the storage to the current thread at 

19 the points where it *intentionally* moves work between threads (see 

20 ``asgiref.sync._restore_context``); data merely inherited by an unrelated 

21 thread is never re-homed and so stays isolated. 

22 """ 

23 

24 __slots__ = ("thread_id", "data") 

25 

26 def __init__(self, thread_id: int, data: dict[str, Any]) -> None: 

27 self.thread_id = thread_id 

28 self.data = data 

29 

30 

31def _rehome(storage: "_Storage") -> "_Storage": 

32 """Return a copy of *storage* owned by the current thread.""" 

33 return _Storage(threading.get_ident(), storage.data) 

34 

35 

36_object_setattr = object.__setattr__ 

37 

38 

39class _CVar: 

40 """Storage utility for Local.""" 

41 

42 _data: "contextvars.ContextVar[_Storage]" 

43 

44 def __init__(self) -> None: 

45 _object_setattr(self, "_data", contextvars.ContextVar("asgiref.local")) 

46 

47 def _storage(self) -> "_Storage": 

48 # Only return storage that belongs to the current thread. Storage with 

49 # a different thread id was inherited by this thread (rather than 

50 # intentionally moved here by asgiref) and must not be visible. 

51 storage = self._data.get(None) 

52 if storage is None or storage.thread_id != threading.get_ident(): 

53 return _Storage(threading.get_ident(), {}) 

54 return storage 

55 

56 def __getattr__(self, key): 

57 try: 

58 return self._storage().data[key] 

59 except KeyError: 

60 raise AttributeError(f"{self!r} object has no attribute {key!r}") 

61 

62 def __setattr__(self, key: str, value: Any) -> None: 

63 if key == "_data": 

64 raise AttributeError("Cannot set attribute '_data' on Local") 

65 

66 data = self._storage().data.copy() 

67 data[key] = value 

68 self._data.set(_Storage(threading.get_ident(), data)) 

69 

70 def __delattr__(self, key: str) -> None: 

71 data = self._storage().data.copy() 

72 if key in data: 

73 del data[key] 

74 self._data.set(_Storage(threading.get_ident(), data)) 

75 else: 

76 raise AttributeError(f"{self!r} object has no attribute {key!r}") 

77 

78 

79@final 

80class Local: 

81 """Local storage for async tasks. 

82 

83 This is a namespace object (similar to `threading.local`) where data is 

84 also local to the current async task (if there is one). 

85 

86 In async threads, local means in the same sense as the `contextvars` 

87 module - i.e. a value set in an async frame will be visible: 

88 

89 - to other async code `await`-ed from this frame. 

90 - to tasks spawned using `asyncio` utilities (`create_task`, `wait_for`, 

91 `gather` and probably others). 

92 - to code scheduled in a sync thread using `sync_to_async` 

93 

94 In "sync" threads (a thread with no async event loop running), the 

95 data is thread-local, but additionally shared with async code executed 

96 via the `async_to_sync` utility, which schedules async code in a new thread 

97 and copies context across to that thread. 

98 

99 If `thread_critical` is True, then the local will only be visible per-thread, 

100 behaving exactly like `threading.local` if the thread is sync, and as 

101 `contextvars` if the thread is async. This allows genuinely thread-sensitive 

102 code (such as DB handles) to be kept strictly to their initial thread and 

103 disable the sharing across `sync_to_async` and `async_to_sync` wrapped calls. 

104 

105 Unlike plain `contextvars` objects, this utility is threadsafe. 

106 """ 

107 

108 def __new__(cls, thread_critical: bool = False) -> "Local": 

109 if cls is not Local: 

110 raise TypeError("asgiref.local.Local cannot be subclassed") 

111 # Make `Local(...)` actually instantiate either of the specialized subclasses. 

112 if thread_critical: 

113 return object.__new__(_ThreadCriticalLocal) 

114 return object.__new__(_DefaultLocal) 

115 

116 def __init__(self, thread_critical: bool = False) -> None: 

117 # `thread_critical` is not really used here; 

118 # this just marks the __init__ interface for 

119 # the concrete specialized subclasses. 

120 pass 

121 

122 # These mark this class as magical to Mypy; the real implementations 

123 # are in the concrete subclasses below. 

124 

125 def __getattr__(self, key: str) -> Any: ... 

126 

127 def __setattr__(self, key: str, value: Any) -> Any: ... 

128 

129 def __delattr__(self, key: str) -> Any: ... 

130 

131 

132class _DefaultLocal(Local): # type: ignore[misc] 

133 """Local with thread_critical=False: CVar storage protected by a lock.""" 

134 

135 def __init__(self, thread_critical: bool = False) -> None: 

136 super().__init__(thread_critical) 

137 storage = _CVar() 

138 _object_setattr(self, "_lock", threading.RLock()) 

139 _object_setattr(self, "_storage", storage) 

140 

141 def __getattr__(self, key): 

142 # Lock acquisition inlined here for performance reasons. 

143 self._lock.acquire() 

144 try: 

145 return getattr(self._storage, key) 

146 finally: 

147 self._lock.release() 

148 

149 def __setattr__(self, key, value): 

150 # Lock acquisition inlined here for performance reasons. 

151 self._lock.acquire() 

152 try: 

153 setattr(self._storage, key, value) 

154 finally: 

155 self._lock.release() 

156 

157 def __delattr__(self, key): 

158 # Lock acquisition inlined here for performance reasons. 

159 self._lock.acquire() 

160 try: 

161 delattr(self._storage, key) 

162 finally: 

163 self._lock.release() 

164 

165 

166class _ThreadCriticalLocal(Local): # type: ignore[misc] 

167 """Local with thread_critical=True: thread-local storage, CVar in async.""" 

168 

169 def __init__(self, thread_critical: bool = False) -> None: 

170 super().__init__(thread_critical) 

171 _object_setattr(self, "_storage", threading.local()) 

172 

173 def _get_storage(self): 

174 try: 

175 asyncio.get_running_loop() 

176 except RuntimeError: 

177 # We are in a sync thread, the storage is 

178 # just the plain thread local (i.e, "global within 

179 # this thread" - it doesn't matter where you are 

180 # in a call stack you see the same storage) 

181 return self._storage 

182 

183 # We are in an async thread - storage is still 

184 # local to this thread, but additionally should 

185 # behave like a context var (is only visible with 

186 # the same async call stack) 

187 

188 # Ensure context exists in the current thread 

189 if not hasattr(self._storage, "cvar"): 

190 self._storage.cvar = _CVar() 

191 

192 # self._storage is a thread local, so the members 

193 # can't be accessed in another thread (we don't 

194 # need any locks) 

195 return self._storage.cvar 

196 

197 def __getattr__(self, key): 

198 return getattr(self._get_storage(), key) 

199 

200 def __setattr__(self, key, value): 

201 setattr(self._get_storage(), key, value) 

202 

203 def __delattr__(self, key): 

204 delattr(self._get_storage(), key)