or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

index.md
tile.json

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

index.mddocs/

Sniffio

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.

Package Information

  • Package Name: sniffio
  • Language: Python
  • Installation: pip install sniffio

Core Imports

import sniffio

Or import specific functions:

from sniffio import current_async_library, AsyncLibraryNotFoundError

Basic Usage

import sniffio
import asyncio
import trio

async def adapt_to_async_library():
    """Example showing how to adapt behavior based on async library."""
    try:
        library = sniffio.current_async_library()
        print(f"Running under: {library}")
        
        if library == "asyncio":
            # Use asyncio-specific functionality
            await asyncio.sleep(1)
        elif library == "trio":
            # Use trio-specific functionality
            await trio.sleep(1)
        elif library == "curio":
            # Use curio-specific functionality
            import curio
            await curio.sleep(1)
        else:
            print(f"Unknown async library: {library}")
            
    except sniffio.AsyncLibraryNotFoundError:
        print("Not running in an async context")

# Run with different async libraries
asyncio.run(adapt_to_async_library())  # Prints "Running under: asyncio" 
trio.run(adapt_to_async_library)       # Prints "Running under: trio"

Capabilities

Async Library Detection

The core functionality for detecting which async library is currently active.

def current_async_library() -> str:
    """
    Detect which async library is currently running.
    
    Supports detection of:
    - Trio (v0.6+): returns "trio"
    - Curio: returns "curio" 
    - asyncio: returns "asyncio"
    - Trio-asyncio (v0.8.2+): returns "trio" or "asyncio" depending on current mode
    
    Returns:
        str: Name of the current async library ("trio", "asyncio", "curio")
        
    Raises:
        AsyncLibraryNotFoundError: If called from synchronous context or if the 
            current async library was not recognized
    """

Manual Library Override

Context variable and thread-local mechanisms for manually setting the detected library.

current_async_library_cvar: ContextVar[Optional[str]]
"""
Context variable for explicitly setting the current async library.
Can be used to override automatic detection.

Usage:
    token = sniffio.current_async_library_cvar.set("custom-lib")
    try:
        library = sniffio.current_async_library()  # Returns "custom-lib"
    finally:
        sniffio.current_async_library_cvar.reset(token)
"""
thread_local: _ThreadLocal
"""
Thread-local storage object for setting async library per thread.
Has a 'name' attribute that can be set to override detection.

Usage:
    old_name = sniffio.thread_local.name
    sniffio.thread_local.name = "custom-lib"
    try:
        library = sniffio.current_async_library()  # Returns "custom-lib"
    finally:
        sniffio.thread_local.name = old_name
"""

Exception Handling

class AsyncLibraryNotFoundError(RuntimeError):
    """
    Exception raised when async library detection fails.
    
    Raised by current_async_library() when:
    - Called from synchronous (non-async) context
    - Current async library is not recognized
    - No async library is running
    """

Types

from typing import Optional
from contextvars import ContextVar
import threading

class _ThreadLocal(threading.local):
    """
    Custom thread-local storage class with default value support.
    
    Attributes:
        name (Optional[str]): Name of the async library for current thread.
            Defaults to None.
    """
    name: Optional[str] = None

Version Information

__version__: str
"""
Package version string. Available as sniffio.__version__ but not included in __all__.
Current version: "1.3.1"
"""

Detection Priority

The library uses a three-tier detection system:

  1. Thread-local storage (thread_local.name) - Highest priority
  2. Context variable (current_async_library_cvar) - Medium priority
  3. Automatic detection - Lowest priority, performed by:
    • Checking for active asyncio task using asyncio.current_task()
    • Checking for curio using curio.meta.curio_running()

This priority system allows manual overrides while providing automatic detection as a fallback.

Usage Examples

Library Detection in Conditional Code

import sniffio

async def generic_sleep(seconds):
    """Sleep function that works with multiple async libraries."""
    library = sniffio.current_async_library()
    
    if library == "trio":
        import trio
        await trio.sleep(seconds)
    elif library == "asyncio":
        import asyncio
        await asyncio.sleep(seconds)
    elif library == "curio":
        import curio
        await curio.sleep(seconds)
    else:
        raise RuntimeError(f"Unsupported library {library!r}")

Manual Override with Context Variables

import sniffio

async def test_with_override():
    # Temporarily override detection
    token = sniffio.current_async_library_cvar.set("custom-lib")
    try:
        library = sniffio.current_async_library()
        assert library == "custom-lib"
    finally:
        sniffio.current_async_library_cvar.reset(token)

Thread-Local Override

import sniffio

def set_thread_library():
    # Set for current thread
    sniffio.thread_local.name = "thread-specific-lib"
    
    # This will return "thread-specific-lib" from any async context in this thread
    # (assuming no context variable override)

Error Handling

import sniffio

def check_async_context():
    """Check if we're running in an async context."""
    try:
        library = sniffio.current_async_library()
        print(f"Async library detected: {library}")
        return True
    except sniffio.AsyncLibraryNotFoundError:
        print("Not in async context")
        return False