Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/sniffio/_impl.py: 33%

1from contextvars import ContextVar 

2from typing import Optional 

3import sys 

4import threading 

5 

6current_async_library_cvar = ContextVar( 

7 "current_async_library_cvar", default=None 

8) # type: ContextVar[Optional[str]] 

9 

10 

11class _ThreadLocal(threading.local): 

12 # Since threading.local provides no explicit mechanism is for setting 

13 # a default for a value, a custom class with a class attribute is used 

14 # instead. 

15 name = None # type: Optional[str] 

16 

17 

18thread_local = _ThreadLocal() 

19 

20 

21class AsyncLibraryNotFoundError(RuntimeError): 

22 pass 

23 

24 

25def current_async_library() -> str: 

26 """Detect which async library is currently running. 

27 

28 The following libraries are currently supported: 

29 

30 ================ =========== ============================ 

31 Library Requires Magic string 

32 ================ =========== ============================ 

33 **Trio** Trio v0.6+ ``"trio"`` 

34 **Curio** - ``"curio"`` 

35 **asyncio** ``"asyncio"`` 

36 **Trio-asyncio** v0.8.2+ ``"trio"`` or ``"asyncio"``, 

37 depending on current mode 

38 ================ =========== ============================ 

39 

40 Returns: 

41 A string like ``"trio"``. 

42 

43 Raises: 

44 AsyncLibraryNotFoundError: if called from synchronous context, 

45 or if the current async library was not recognized. 

46 

47 Examples: 

48 

49 .. code-block:: python3 

50 

51 from sniffio import current_async_library 

52 

53 async def generic_sleep(seconds): 

54 library = current_async_library() 

55 if library == "trio": 

56 import trio 

57 await trio.sleep(seconds) 

58 elif library == "asyncio": 

59 import asyncio 

60 await asyncio.sleep(seconds) 

61 # ... and so on ... 

62 else: 

63 raise RuntimeError(f"Unsupported library {library!r}") 

64 

65 """ 

66 value = thread_local.name 

67 if value is not None: 

68 return value 

69 

70 value = current_async_library_cvar.get() 

71 if value is not None: 

72 return value 

73 

74 # Need to sniff for asyncio 

75 if "asyncio" in sys.modules: 

76 import asyncio 

77 try: 

78 current_task = asyncio.current_task # type: ignore[attr-defined] 

79 except AttributeError: 

80 current_task = asyncio.Task.current_task # type: ignore[attr-defined] 

81 try: 

82 if current_task() is not None: 

83 return "asyncio" 

84 except RuntimeError: 

85 pass 

86 

87 # Sniff for curio (for now) 

88 if 'curio' in sys.modules: 

89 from curio.meta import curio_running 

90 if curio_running(): 

91 return 'curio' 

92 

93 raise AsyncLibraryNotFoundError( 

94 "unknown async library, or not in async context" 

95 )