or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

tessl/pypi-sniffio

Sniff out which async library your code is running under

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/sniffio@1.3.x

To install, run

npx @tessl/cli install tessl/pypi-sniffio@1.3.0

0

# 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

```