Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Synchronization decorators and calling-convention markers/bridges.
3Provides ``synchronized`` for thread and async locking, ``mark_as_sync``
4and ``mark_as_async`` for declaring the effective calling convention of a
5wrapped callable (without converting it), and ``async_to_sync`` /
6``sync_to_async`` for bridging between the two.
7"""
9import asyncio
10import sys
11import warnings
12from functools import partial
13from inspect import (
14 CO_ASYNC_GENERATOR,
15 CO_COROUTINE,
16 CO_GENERATOR,
17 CO_ITERABLE_COROUTINE,
18 isasyncgenfunction,
19 iscoroutinefunction,
20 isgeneratorfunction,
21)
22from threading import Lock, RLock
24from .__wrapt__ import BoundFunctionWrapper, CallableObjectProxy, FunctionWrapper
25from .decorators import decorator
27# Calling-convention marker wrappers. These manipulate __code__.co_flags
28# so that inspect.iscoroutinefunction() reports the intended calling
29# convention, which lets stdlib code and the synchronized() decorator
30# auto-select the correct sync or async wrapping behaviour even when
31# stacked decorators change the effective convention (for example an
32# inner decorator that invokes an async def via asyncio.run()).
35class _SyncCodeProxy(CallableObjectProxy):
37 def __init__(self, wrapped, generator=None):
38 super().__init__(wrapped)
39 self._self_generator = generator
41 @property
42 def co_flags(self):
43 original = self.__wrapped__.co_flags
44 # Strip async-axis and iterable-coroutine bits; sync means neither
45 # coroutine function nor async generator nor types.coroutine-style.
46 flags = original & ~(CO_COROUTINE | CO_ASYNC_GENERATOR | CO_ITERABLE_COROUTINE)
47 if self._self_generator is True:
48 flags |= CO_GENERATOR
49 elif self._self_generator is False:
50 flags &= ~CO_GENERATOR
51 else:
52 # Auto: if input was an async generator, preserve generator-ness
53 # on the sync side by setting CO_GENERATOR. Otherwise leave
54 # CO_GENERATOR as-is (already copied from the wrapped flags).
55 if original & CO_ASYNC_GENERATOR:
56 flags |= CO_GENERATOR
57 return flags
60class _SyncFunctionSurrogate(CallableObjectProxy):
62 def __init__(self, wrapped, generator=None):
63 super().__init__(wrapped)
64 self._self_generator = generator
66 @property
67 def __code__(self):
68 return _SyncCodeProxy(self.__wrapped__.__code__, self._self_generator)
71class _BoundSyncFunctionWrapper(BoundFunctionWrapper):
73 def __init__(self, *args, **kwargs):
74 super().__init__(*args, **kwargs)
75 self._self_is_not_coroutine = True
77 @property
78 def __func__(self):
79 return _SyncFunctionSurrogate(
80 self.__wrapped__.__func__, self._self_parent._self_generator
81 )
84class _SyncFunctionWrapper(FunctionWrapper):
86 __bound_function_wrapper__ = _BoundSyncFunctionWrapper
88 def __init__(self, wrapped, wrapper, generator=None):
89 super().__init__(wrapped, wrapper)
90 self._self_is_not_coroutine = True
91 self._self_generator = generator
93 @property
94 def __code__(self):
95 return _SyncCodeProxy(self.__wrapped__.__code__, self._self_generator)
98class _AsyncCodeProxy(CallableObjectProxy):
100 def __init__(self, wrapped, generator=None):
101 super().__init__(wrapped)
102 self._self_generator = generator
104 @property
105 def co_flags(self):
106 original = self.__wrapped__.co_flags
107 # Strip all four convention bits; we reassert the right ones below.
108 flags = original & ~(
109 CO_GENERATOR | CO_COROUTINE | CO_ITERABLE_COROUTINE | CO_ASYNC_GENERATOR
110 )
111 if self._self_generator is True:
112 flags |= CO_ASYNC_GENERATOR
113 elif self._self_generator is False:
114 flags |= CO_COROUTINE
115 else:
116 # Auto: if input was a generator (sync or async), produce an
117 # async generator; otherwise produce a coroutine function.
118 if original & (CO_GENERATOR | CO_ASYNC_GENERATOR):
119 flags |= CO_ASYNC_GENERATOR
120 else:
121 flags |= CO_COROUTINE
122 return flags
125class _AsyncFunctionSurrogate(CallableObjectProxy):
127 def __init__(self, wrapped, generator=None):
128 super().__init__(wrapped)
129 self._self_generator = generator
131 @property
132 def __code__(self):
133 return _AsyncCodeProxy(self.__wrapped__.__code__, self._self_generator)
136class _BoundAsyncFunctionWrapper(BoundFunctionWrapper):
138 @property
139 def __func__(self):
140 return _AsyncFunctionSurrogate(
141 self.__wrapped__.__func__, self._self_parent._self_generator
142 )
145class _AsyncFunctionWrapper(FunctionWrapper):
147 __bound_function_wrapper__ = _BoundAsyncFunctionWrapper
149 def __init__(self, wrapped, wrapper, generator=None):
150 super().__init__(wrapped, wrapper)
151 self._self_generator = generator
153 @property
154 def __code__(self):
155 return _AsyncCodeProxy(self.__wrapped__.__code__, self._self_generator)
158def mark_as_sync(wrapped=None, /, *, generator=None):
159 """Mark a callable as synchronous from the perspective of calling
160 convention detection. The returned wrapper is a pass-through that
161 reports `inspect.iscoroutinefunction()` as False regardless of
162 whether the underlying callable is declared `async def`. Useful
163 when a stacked decorator has already collapsed an async function
164 into a synchronous one (for example by using `asyncio.run()`).
166 The `generator` keyword toggles the sync generator bit
167 (`CO_GENERATOR`) on the resulting wrapper. Tri-state:
169 - `None` (default): auto. Preserve generator-ness from the input --
170 if the input was an async generator, the wrapper reports as a sync
171 generator; otherwise CO_GENERATOR is copied through unchanged.
172 - `True`: force CO_GENERATOR on. Wrapper reports as a sync generator.
173 - `False`: force CO_GENERATOR off. Wrapper reports as a plain sync
174 function even if the input had CO_GENERATOR set.
176 Regardless of `generator`, CO_COROUTINE, CO_ASYNC_GENERATOR, and
177 CO_ITERABLE_COROUTINE are all cleared (sync means none of those).
178 """
180 def _decorator(wrapped):
181 def _wrapper(wrapped, instance, args, kwargs):
182 return wrapped(*args, **kwargs)
184 return _SyncFunctionWrapper(wrapped, _wrapper, generator=generator)
186 if wrapped is None:
187 return _decorator
188 return _decorator(wrapped)
191def mark_as_async(wrapped=None, /, *, generator=None):
192 """Mark a callable as asynchronous from the perspective of calling
193 convention detection. The returned wrapper reports
194 `inspect.iscoroutinefunction()` as True regardless of whether the
195 underlying callable is declared `async def`. Useful when a stacked
196 decorator returns a coroutine from a plain `def` wrapper.
198 The `generator` keyword chooses between coroutine function and
199 async generator reporting. Tri-state:
201 - `None` (default): auto. If the input was a sync or async
202 generator, the wrapper reports as an async generator
203 (`CO_ASYNC_GENERATOR`); otherwise it reports as a coroutine
204 function (`CO_COROUTINE`).
205 - `True`: force async generator reporting (`CO_ASYNC_GENERATOR` set,
206 `CO_COROUTINE` cleared). These two flags are mutually exclusive at
207 the CPython code-object level.
208 - `False`: force coroutine function reporting (`CO_COROUTINE` set,
209 `CO_ASYNC_GENERATOR` cleared).
211 CO_GENERATOR and CO_ITERABLE_COROUTINE are always cleared (the
212 async path does not use either).
213 """
215 def _decorator(wrapped):
216 def _wrapper(wrapped, instance, args, kwargs):
217 return wrapped(*args, **kwargs)
219 return _AsyncFunctionWrapper(wrapped, _wrapper, generator=generator)
221 if wrapped is None:
222 return _decorator
223 return _decorator(wrapped)
226def async_to_sync(wrapped):
227 """Adapt an async callable so it can be called synchronously. Each
228 call runs the coroutine to completion via `asyncio.run()`. The
229 returned wrapper reports as synchronous under
230 `inspect.iscoroutinefunction()`. Naming follows the asgiref
231 convention.
233 Generator functions of either convention are rejected with a
234 `TypeError`: `asyncio.run()` cannot run an async generator, and a
235 sync generator has nothing to run to completion. A callable not
236 reporting as a coroutine function produces a `UserWarning` rather
237 than an error, since convention detection has false negatives such
238 as a plain function wrapper which returns a coroutine; if the
239 callable really is asynchronous in behaviour, apply
240 `mark_as_async` to it first so it reports as such."""
242 if isasyncgenfunction(wrapped):
243 raise TypeError(
244 "async_to_sync cannot be applied to an async generator "
245 "function; asyncio.run() cannot run an async generator."
246 )
248 if isgeneratorfunction(wrapped):
249 raise TypeError(
250 "async_to_sync cannot be applied to a generator function; "
251 "there is no coroutine to run to completion. If the callable "
252 "is misreported and actually asynchronous in behaviour, apply "
253 "mark_as_async to it first."
254 )
256 if not iscoroutinefunction(wrapped):
257 # Detection has false negatives, such as a plain def wrapper
258 # which returns a coroutine, so this cannot be an error. Such a
259 # callable works here regardless, but marking it first also
260 # corrects what introspection reports elsewhere.
261 warnings.warn(
262 "async_to_sync was applied to a callable not reporting as a "
263 "coroutine function. If it is asynchronous in behaviour, "
264 "apply mark_as_async to it first so it reports as such.",
265 UserWarning,
266 stacklevel=2,
267 )
269 def wrapper(wrapped, instance, args, kwargs):
270 return asyncio.run(wrapped(*args, **kwargs))
272 return _SyncFunctionWrapper(wrapped, wrapper, generator=False)
275def sync_to_async(wrapped):
276 """Adapt a sync callable so it can be awaited. Each call dispatches
277 the synchronous work to the default executor via
278 `loop.run_in_executor()`. The returned wrapper reports as
279 asynchronous under `inspect.iscoroutinefunction()`. Naming follows
280 the asgiref convention.
282 Callables reporting as coroutine functions or generator functions
283 of either convention are rejected with a `TypeError`. For a
284 generator function the adaptation would be an illusion: only
285 creation of the generator would be dispatched to the executor,
286 which executes no body code, while each iteration would still block
287 the event loop. A callable misreported as asynchronous but
288 synchronous in behaviour can have `mark_as_sync` applied to it
289 first so it reports as such."""
291 if iscoroutinefunction(wrapped) or isasyncgenfunction(wrapped):
292 # Mirrors the equivalent check in asgiref. A callable which is
293 # misreported, such as a wrapper which runs an inner async def
294 # to completion, can be corrected with mark_as_sync first.
295 raise TypeError(
296 "sync_to_async can only be applied to a synchronous "
297 "callable. If the callable is misreported and actually "
298 "synchronous in behaviour, apply mark_as_sync to it first."
299 )
301 if isgeneratorfunction(wrapped):
302 raise TypeError(
303 "sync_to_async cannot be applied to a generator function; "
304 "creating the generator executes no code, and iterating it "
305 "would still block the event loop."
306 )
308 async def wrapper(wrapped, instance, args, kwargs):
309 loop = asyncio.get_running_loop()
310 return await loop.run_in_executor(None, partial(wrapped, *args, **kwargs))
312 return _AsyncFunctionWrapper(wrapped, wrapper, generator=False)
315def _synchronized_is_async_lock(obj):
316 return iscoroutinefunction(getattr(obj, "acquire", None))
319def _synchronized_is_async_callable(obj):
320 # Walk the __wrapped__ chain, returning True as soon as any layer
321 # declares itself a coroutine function. A sync marker wrapper can
322 # carry an authoritative `_self_is_not_coroutine` attribute that
323 # short-circuits the walk before it descends into a genuinely
324 # async inner layer. Cycle / runaway-chain protection modelled on
325 # inspect.unwrap().
327 memo = {id(obj): obj}
328 recursion_limit = sys.getrecursionlimit()
329 target = obj
331 while True:
332 if isinstance(target, (classmethod, staticmethod)):
333 inner = getattr(target, "__wrapped__", None)
334 if inner is None:
335 inner = target.__func__
336 target = inner
337 id_target = id(target)
338 if id_target in memo or len(memo) >= recursion_limit:
339 raise ValueError("wrapper loop when unwrapping {!r}".format(obj))
340 memo[id_target] = target
341 continue
343 if getattr(target, "_self_is_not_coroutine", False):
344 return False
346 if iscoroutinefunction(target):
347 return True
349 next_target = getattr(target, "__wrapped__", None)
350 if next_target is None or next_target is target:
351 return False
352 target = next_target
353 id_target = id(target)
354 if id_target in memo or len(memo) >= recursion_limit:
355 raise ValueError("wrapper loop when unwrapping {!r}".format(obj))
356 memo[id_target] = target
359# Decorator for implementing thread synchronization. It can be used as a
360# decorator, in which case the synchronization context is determined by
361# what type of function is wrapped, or it can also be used as a context
362# manager, where the user needs to supply the correct synchronization
363# context. It is also possible to supply an object which appears to be a
364# synchronization primitive of some sort, by virtue of having release()
365# and acquire() methods. In that case that will be used directly as the
366# synchronization primitive without creating a separate lock against the
367# derived or supplied context.
370def synchronized(wrapped):
371 """Depending on the nature of the `wrapped` object, will either return a
372 decorator which can be used to wrap a function or method, or a context
373 manager, both of which will act accordingly depending on how used, to
374 synchronize access to calling of the wrapped function, or the block of
375 code within the context manager. If it is an object which is a
376 synchronization primitive, such as a threading Lock, RLock, Semaphore,
377 Condition, or Event, then it is assumed that the object is to be used
378 directly as the synchronization primitive, otherwise a lock is created
379 automatically and attached to the wrapped object and used as the
380 synchronization primitive.
382 Async functions are supported: if the wrapped callable is an async
383 function, an `asyncio.Lock` is created for the context and the wrapper
384 awaits the lock. The returned object also exposes `__aenter__` and
385 `__aexit__` so it can be used with `async with` to synchronise a block
386 of code using an independent per-context `asyncio.Lock`. If an object
387 with coroutine `acquire`/`release` methods (such as an `asyncio.Lock`)
388 is supplied directly, the returned decorator and context manager will
389 use it via the async protocol.
390 """
392 # Determine if being passed an object which is a synchronization
393 # primitive. We can't check by type for Lock, RLock, Semaphore etc,
394 # as the means of creating them isn't the type. Therefore use the
395 # existence of acquire() and release() methods. This is more
396 # extensible anyway as it allows custom synchronization mechanisms.
398 if hasattr(wrapped, "acquire") and hasattr(wrapped, "release"):
399 # We remember what the original lock is and then return a new
400 # decorator which accesses and locks it. When returning the new
401 # decorator we wrap it with an object proxy so we can override
402 # the context manager methods in case it is being used to wrap
403 # synchronized statements with a 'with' statement.
405 lock = wrapped
407 if _synchronized_is_async_lock(lock):
409 @decorator
410 async def _synchronized(wrapped, instance, args, kwargs):
411 async with lock:
412 return await wrapped(*args, **kwargs)
414 class _AsyncSynchronizedLockProxy(CallableObjectProxy):
416 async def __aenter__(self):
417 await lock.acquire()
418 return lock
420 async def __aexit__(self, *args):
421 lock.release()
423 return _AsyncSynchronizedLockProxy(wrapped=_synchronized)
425 @decorator
426 def _synchronized(wrapped, instance, args, kwargs):
427 # Execute the wrapped function while the original supplied
428 # lock is held.
430 with lock:
431 return wrapped(*args, **kwargs)
433 class _SynchronizedLockProxy(CallableObjectProxy):
435 def __enter__(self):
436 lock.acquire()
437 return lock
439 def __exit__(self, *args):
440 lock.release()
442 return _SynchronizedLockProxy(wrapped=_synchronized)
444 # Following only apply when the lock is being created automatically
445 # based on the context of what was supplied. In this case we supply
446 # a final decorator, but need to use FunctionWrapper directly as we
447 # want to derive from it to add context manager methods in case it is
448 # being used to wrap synchronized statements with a 'with' statement.
450 def _synchronized_lock(context):
451 # Attempt to retrieve the lock for the specific context.
453 lock = vars(context).get("_synchronized_lock", None)
455 if lock is None:
456 # There is no existing lock defined for the context we
457 # are dealing with so we need to create one. This needs
458 # to be done in a way to guarantee there is only one
459 # created, even if multiple threads try and create it at
460 # the same time. We can't always use the setdefault()
461 # method on the __dict__ for the context. This is the
462 # case where the context is a class, as __dict__ is
463 # actually a dictproxy. What we therefore do is use a
464 # meta lock on this wrapper itself, to control the
465 # creation and assignment of the lock attribute against
466 # the context.
468 with synchronized._synchronized_meta_lock:
469 # We need to check again for whether the lock we want
470 # exists in case two threads were trying to create it
471 # at the same time and were competing to create the
472 # meta lock.
474 lock = vars(context).get("_synchronized_lock", None)
476 if lock is None:
477 lock = RLock()
478 setattr(context, "_synchronized_lock", lock)
480 return lock
482 def _synchronized_async_lock(context):
483 # Per-context asyncio.Lock, created lazily on first use. Created
484 # under the shared meta lock so creation is safe across threads;
485 # the meta lock is never held across an await. asyncio.Lock is
486 # not reentrant.
488 lock = vars(context).get("_synchronized_async_lock", None)
490 if lock is None:
491 with synchronized._synchronized_meta_lock:
492 lock = vars(context).get("_synchronized_async_lock", None)
494 if lock is None:
495 lock = asyncio.Lock()
496 setattr(context, "_synchronized_async_lock", lock)
498 return lock
500 def _synchronized_wrapper(wrapped, instance, args, kwargs):
501 # Execute the wrapped function while the lock for the
502 # desired context is held. If instance is None then the
503 # wrapped function is used as the context.
505 with _synchronized_lock(instance if instance is not None else wrapped):
506 return wrapped(*args, **kwargs)
508 async def _synchronized_async_wrapper(wrapped, instance, args, kwargs):
509 async with _synchronized_async_lock(
510 instance if instance is not None else wrapped
511 ):
512 return await wrapped(*args, **kwargs)
514 class _SynchronizedFunctionWrapper(FunctionWrapper):
516 def __enter__(self):
517 self._self_lock = _synchronized_lock(self.__wrapped__)
518 self._self_lock.acquire()
519 return self._self_lock
521 def __exit__(self, *args):
522 self._self_lock.release()
524 async def __aenter__(self):
525 self._self_async_lock = _synchronized_async_lock(self.__wrapped__)
526 await self._self_async_lock.acquire()
527 return self._self_async_lock
529 async def __aexit__(self, *args):
530 self._self_async_lock.release()
532 if _synchronized_is_async_callable(wrapped):
533 return _SynchronizedFunctionWrapper(
534 wrapped=wrapped, wrapper=_synchronized_async_wrapper
535 )
537 return _SynchronizedFunctionWrapper(wrapped=wrapped, wrapper=_synchronized_wrapper)
540synchronized._synchronized_meta_lock = Lock() # type: ignore[attr-defined]