asgiref

ASGI specs, helper code, and adapters for bridging synchronous and asynchronous Python web applications. This library provides essential utilities for asynchronous web development, serving as the foundational component for Django Channels and other ASGI-based applications.

Package Information

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

Core Imports

import asgiref

Most commonly used imports:

from asgiref.sync import sync_to_async, async_to_sync
from asgiref.wsgi import WsgiToAsgi
from asgiref.compatibility import guarantee_single_callable

Basic Usage

from asgiref.sync import sync_to_async, async_to_sync
import asyncio

# Convert sync function to async
@sync_to_async
def sync_database_query():
    # Simulate database query
    return {"id": 1, "name": "example"}

# Convert async function to sync
@async_to_sync
async def async_api_call():
    # Simulate async API call
    await asyncio.sleep(0.1)
    return {"status": "success"}

# Usage
async def main():
    # Call sync function from async context
    result = await sync_database_query()
    print(result)

# Call async function from sync context
result = async_api_call()
print(result)

Architecture

asgiref's design enables seamless integration between synchronous and asynchronous Python code:

• Sync/Async Converters: Core utilities that handle context propagation and thread management
• ASGI Compatibility Layer: Ensures applications work with both ASGI2 and ASGI3 specifications
• Thread-Safe Storage: Async-aware replacements for threading.local that work across event loops
• WSGI Integration: Adapters for running legacy WSGI applications in ASGI servers
• Testing Utilities: Tools for testing ASGI applications in controlled environments

This architecture supports Django's transition to async views while maintaining backward compatibility with synchronous components.

Capabilities

Sync/Async Conversion

Core functionality for converting between synchronous and asynchronous functions, handling context propagation, thread management, and event loop integration automatically.

def sync_to_async(func=None, *, thread_sensitive=True, executor=None): ...
def async_to_sync(awaitable=None, *, force_new_loop=False): ...
class SyncToAsync: ...
class AsyncToSync: ...

Sync/Async Conversion

ASGI Compatibility

Utilities for ensuring ASGI applications work consistently across different ASGI versions and server implementations, handling legacy double-callable patterns.

def guarantee_single_callable(application): ...
def is_double_callable(application): ...
def double_to_single_callable(application): ...

ASGI Compatibility

WSGI Integration

Adapters for running WSGI applications within ASGI servers, enabling legacy web applications to benefit from async server capabilities.

class WsgiToAsgi:
    def __init__(wsgi_application): ...
    def __call__(scope, receive, send): ...

WSGI Integration

Thread-Safe Storage

Async-aware local storage that works correctly with both threading and asyncio, replacing threading.local for async applications.

class Local:
    def __init__(thread_critical=False): ...

Thread-Safe Storage

ASGI Server Base

Base classes for implementing ASGI servers, handling application lifecycle, connection management, and stateless protocol support.

class StatelessServer:
    def __init__(application, max_applications=1000): ...
    def run(): ...
    async def arun(): ...

ASGI Server Base

Testing Utilities

Tools for testing ASGI applications, providing controlled communication channels and application lifecycle management.

class ApplicationCommunicator:
    def __init__(application, scope): ...
    async def wait(timeout=1): ...
    async def send_input(message): ...
    async def receive_output(timeout=1): ...

Testing Utilities

Timeout Management

Async timeout context managers for controlling operation timeouts in asynchronous code.

class timeout:
    def __init__(timeout, *, loop=None): ...
    async def __aenter__(): ...
    async def __aexit__(exc_type, exc_val, exc_tb): ...

Timeout Management

Current Thread Executor

Specialized executor for running code in the thread where it was instantiated, enabling cross-thread execution coordination for async-to-sync conversion scenarios.

class CurrentThreadExecutor:
    def __init__(self, old_executor: CurrentThreadExecutor | None) -> None: ...
    def run_until_future(self, future: Future[Any]) -> None: ...
    def submit(self, fn: Callable[_P, _R], /, *args: _P.args, **kwargs: _P.kwargs) -> Future[_R]: ...

Current Thread Executor

Type Definitions

Complete type definitions for ASGI protocols, events, and applications, enabling full type safety in async web applications.

# Protocol Types
ASGIApplication = Union[ASGI2Application, ASGI3Application]
ASGIReceiveCallable = Callable[[], Awaitable[ASGIReceiveEvent]]
ASGISendCallable = Callable[[ASGISendEvent], Awaitable[None]]

# Scope Types  
Scope = Union[HTTPScope, WebSocketScope, LifespanScope]

Type Definitions

Version Information

__version__: str  # Package version "3.9.1"