Sniff out which async library your code is running under
npx @tessl/cli install tessl/pypi-sniffio@1.3.00
# Sniffio
1
2
A lightweight Python library that detects which asynchronous I/O library is currently running in your code execution context. Sniffio enables library developers to write code that adapts its behavior based on the async runtime environment, supporting popular frameworks including Trio, asyncio, and Curio.
3
4
## Package Information
5
6
- **Package Name**: sniffio
7
- **Language**: Python
8
- **Installation**: `pip install sniffio`
9
10
## Core Imports
11
12
```python
13
import sniffio
14
```
15
16
Or import specific functions:
17
18
```python
19
from sniffio import current_async_library, AsyncLibraryNotFoundError
20
```
21
22
## Basic Usage
23
24
```python
25
import sniffio
26
import asyncio
27
import trio
28
29
async def adapt_to_async_library():
30
"""Example showing how to adapt behavior based on async library."""
31
try:
32
library = sniffio.current_async_library()
33
print(f"Running under: {library}")
34
35
if library == "asyncio":
36
# Use asyncio-specific functionality
37
await asyncio.sleep(1)
38
elif library == "trio":
39
# Use trio-specific functionality
40
await trio.sleep(1)
41
elif library == "curio":
42
# Use curio-specific functionality
43
import curio
44
await curio.sleep(1)
45
else:
46
print(f"Unknown async library: {library}")
47
48
except sniffio.AsyncLibraryNotFoundError:
49
print("Not running in an async context")
50
51
# Run with different async libraries
52
asyncio.run(adapt_to_async_library()) # Prints "Running under: asyncio"
53
trio.run(adapt_to_async_library) # Prints "Running under: trio"
54
```
55
56
## Capabilities
57
58
### Async Library Detection
59
60
The core functionality for detecting which async library is currently active.
61
62
```python { .api }
63
def current_async_library() -> str:
64
"""
65
Detect which async library is currently running.
66
67
Supports detection of:
68
- Trio (v0.6+): returns "trio"
69
- Curio: returns "curio"
70
- asyncio: returns "asyncio"
71
- Trio-asyncio (v0.8.2+): returns "trio" or "asyncio" depending on current mode
72
73
Returns:
74
str: Name of the current async library ("trio", "asyncio", "curio")
75
76
Raises:
77
AsyncLibraryNotFoundError: If called from synchronous context or if the
78
current async library was not recognized
79
"""
80
```
81
82
### Manual Library Override
83
84
Context variable and thread-local mechanisms for manually setting the detected library.
85
86
```python { .api }
87
current_async_library_cvar: ContextVar[Optional[str]]
88
"""
89
Context variable for explicitly setting the current async library.
90
Can be used to override automatic detection.
91
92
Usage:
93
token = sniffio.current_async_library_cvar.set("custom-lib")
94
try:
95
library = sniffio.current_async_library() # Returns "custom-lib"
96
finally:
97
sniffio.current_async_library_cvar.reset(token)
98
"""
99
```
100
101
```python { .api }
102
thread_local: _ThreadLocal
103
"""
104
Thread-local storage object for setting async library per thread.
105
Has a 'name' attribute that can be set to override detection.
106
107
Usage:
108
old_name = sniffio.thread_local.name
109
sniffio.thread_local.name = "custom-lib"
110
try:
111
library = sniffio.current_async_library() # Returns "custom-lib"
112
finally:
113
sniffio.thread_local.name = old_name
114
"""
115
```
116
117
### Exception Handling
118
119
```python { .api }
120
class AsyncLibraryNotFoundError(RuntimeError):
121
"""
122
Exception raised when async library detection fails.
123
124
Raised by current_async_library() when:
125
- Called from synchronous (non-async) context
126
- Current async library is not recognized
127
- No async library is running
128
"""
129
```
130
131
## Types
132
133
```python { .api }
134
from typing import Optional
135
from contextvars import ContextVar
136
import threading
137
138
class _ThreadLocal(threading.local):
139
"""
140
Custom thread-local storage class with default value support.
141
142
Attributes:
143
name (Optional[str]): Name of the async library for current thread.
144
Defaults to None.
145
"""
146
name: Optional[str] = None
147
```
148
149
## Version Information
150
151
```python { .api }
152
__version__: str
153
"""
154
Package version string. Available as sniffio.__version__ but not included in __all__.
155
Current version: "1.3.1"
156
"""
157
```
158
159
## Detection Priority
160
161
The library uses a three-tier detection system:
162
163
1. **Thread-local storage** (`thread_local.name`) - Highest priority
164
2. **Context variable** (`current_async_library_cvar`) - Medium priority
165
3. **Automatic detection** - Lowest priority, performed by:
166
- Checking for active asyncio task using `asyncio.current_task()`
167
- Checking for curio using `curio.meta.curio_running()`
168
169
This priority system allows manual overrides while providing automatic detection as a fallback.
170
171
## Usage Examples
172
173
### Library Detection in Conditional Code
174
175
```python
176
import sniffio
177
178
async def generic_sleep(seconds):
179
"""Sleep function that works with multiple async libraries."""
180
library = sniffio.current_async_library()
181
182
if library == "trio":
183
import trio
184
await trio.sleep(seconds)
185
elif library == "asyncio":
186
import asyncio
187
await asyncio.sleep(seconds)
188
elif library == "curio":
189
import curio
190
await curio.sleep(seconds)
191
else:
192
raise RuntimeError(f"Unsupported library {library!r}")
193
```
194
195
### Manual Override with Context Variables
196
197
```python
198
import sniffio
199
200
async def test_with_override():
201
# Temporarily override detection
202
token = sniffio.current_async_library_cvar.set("custom-lib")
203
try:
204
library = sniffio.current_async_library()
205
assert library == "custom-lib"
206
finally:
207
sniffio.current_async_library_cvar.reset(token)
208
```
209
210
### Thread-Local Override
211
212
```python
213
import sniffio
214
215
def set_thread_library():
216
# Set for current thread
217
sniffio.thread_local.name = "thread-specific-lib"
218
219
# This will return "thread-specific-lib" from any async context in this thread
220
# (assuming no context variable override)
221
```
222
223
### Error Handling
224
225
```python
226
import sniffio
227
228
def check_async_context():
229
"""Check if we're running in an async context."""
230
try:
231
library = sniffio.current_async_library()
232
print(f"Async library detected: {library}")
233
return True
234
except sniffio.AsyncLibraryNotFoundError:
235
print("Not in async context")
236
return False
237
```