blob: c1a7bbf218ba985b87cd1d9b23da69222894c1dd (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
|
from contextvars import ContextVar
from typing import Optional
import sys
import threading
current_async_library_cvar = ContextVar(
"current_async_library_cvar", default=None
) # type: ContextVar[Optional[str]]
class _ThreadLocal(threading.local):
# Since threading.local provides no explicit mechanism is for setting
# a default for a value, a custom class with a class attribute is used
# instead.
name = None # type: Optional[str]
thread_local = _ThreadLocal()
class AsyncLibraryNotFoundError(RuntimeError):
pass
def current_async_library() -> str:
"""Detect which async library is currently running.
The following libraries are currently supported:
================ =========== ============================
Library Requires Magic string
================ =========== ============================
**Trio** Trio v0.6+ ``"trio"``
**Curio** - ``"curio"``
**asyncio** ``"asyncio"``
**Trio-asyncio** v0.8.2+ ``"trio"`` or ``"asyncio"``,
depending on current mode
================ =========== ============================
Returns:
A string like ``"trio"``.
Raises:
AsyncLibraryNotFoundError: if called from synchronous context,
or if the current async library was not recognized.
Examples:
.. code-block:: python3
from sniffio import current_async_library
async def generic_sleep(seconds):
library = current_async_library()
if library == "trio":
import trio
await trio.sleep(seconds)
elif library == "asyncio":
import asyncio
await asyncio.sleep(seconds)
# ... and so on ...
else:
raise RuntimeError(f"Unsupported library {library!r}")
"""
value = thread_local.name
if value is not None:
return value
value = current_async_library_cvar.get()
if value is not None:
return value
# Need to sniff for asyncio
if "asyncio" in sys.modules:
import asyncio
try:
current_task = asyncio.current_task # type: ignore[attr-defined]
except AttributeError:
current_task = asyncio.Task.current_task # type: ignore[attr-defined]
try:
if current_task() is not None:
return "asyncio"
except RuntimeError:
pass
# Sniff for curio (for now)
if 'curio' in sys.modules:
from curio.meta import curio_running
if curio_running():
return 'curio'
raise AsyncLibraryNotFoundError(
"unknown async library, or not in async context"
)
|
