CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-asgiref

ASGI specs, helper code, and adapters for bridging synchronous and asynchronous Python web applications

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview
Eval results
Files

type-definitions.mddocs/

Type Definitions

Complete type definitions for ASGI protocols, events, and applications, enabling full type safety in async web applications. This module provides comprehensive TypeScript-style type annotations for all ASGI components.

Capabilities

Protocol Types

Core ASGI application and protocol type definitions that define the interface contracts for ASGI servers and applications.

# Application Types
ASGIApplication = Union[ASGI2Application, ASGI3Application]
ASGI2Application = Callable[[Scope], ASGI2Protocol]
ASGI3Application = Callable[[Scope, ASGIReceiveCallable, ASGISendCallable], Awaitable[None]]
ASGI2Protocol = Callable[[ASGIReceiveCallable, ASGISendCallable], Awaitable[None]]

# Channel Types  
ASGIReceiveCallable = Callable[[], Awaitable[ASGIReceiveEvent]]
ASGISendCallable = Callable[[ASGISendEvent], Awaitable[None]]

# Event Types
ASGIReceiveEvent = Union[HTTPRequestEvent, HTTPDisconnectEvent, WebSocketConnectEvent, WebSocketReceiveEvent, WebSocketDisconnectEvent, LifespanStartupEvent, LifespanShutdownEvent]
ASGISendEvent = Union[HTTPResponseStartEvent, HTTPResponseBodyEvent, HTTPResponseTrailersEvent, HTTPServerPushEvent, WebSocketAcceptEvent, WebSocketSendEvent, WebSocketCloseEvent, LifespanStartupCompleteEvent, LifespanStartupFailedEvent, LifespanShutdownCompleteEvent, LifespanShutdownFailedEvent]

Scope Types

ASGI scope definitions that contain request/connection information passed to applications.

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

# HTTP Scope
HTTPScope = TypedDict('HTTPScope', {
    'type': Literal['http'],
    'asgi': ASGIVersions,
    'http_version': str,
    'method': str,
    'scheme': str,
    'path': str,
    'raw_path': bytes,
    'query_string': bytes,
    'root_path': str,
    'headers': Iterable[Tuple[bytes, bytes]],
    'client': Optional[Tuple[str, int]],
    'server': Optional[Tuple[str, Optional[int]]],
    'state': NotRequired[Dict[str, Any]],
    'extensions': Optional[Dict[str, Dict[object, object]]]
})

# WebSocket Scope  
WebSocketScope = TypedDict('WebSocketScope', {
    'type': Literal['websocket'],
    'asgi': ASGIVersions,
    'http_version': str,
    'scheme': str,
    'path': str,
    'raw_path': bytes,
    'query_string': bytes,
    'root_path': str,
    'headers': Iterable[Tuple[bytes, bytes]],
    'client': Optional[Tuple[str, int]],
    'server': Optional[Tuple[str, Optional[int]]],
    'subprotocols': Iterable[str],
    'state': NotRequired[Dict[str, Any]],
    'extensions': Optional[Dict[str, Dict[object, object]]]
})

# Lifespan Scope
LifespanScope = TypedDict('LifespanScope', {
    'type': Literal['lifespan'],
    'asgi': ASGIVersions,
    'state': NotRequired[Dict[str, Any]],
    'extensions': Optional[Dict[str, Dict[object, object]]]
})

# Legacy WWW Scope (deprecated)
WWWScope = Union[HTTPScope, WebSocketScope]

HTTP Event Types

Type definitions for HTTP protocol events exchanged between servers and applications.

# HTTP Request Events
HTTPRequestEvent = TypedDict('HTTPRequestEvent', {
    'type': Literal['http.request'],
    'body': bytes,
    'more_body': bool
})

HTTPDisconnectEvent = TypedDict('HTTPDisconnectEvent', {
    'type': Literal['http.disconnect']
})

# HTTP Response Events
HTTPResponseStartEvent = TypedDict('HTTPResponseStartEvent', {
    'type': Literal['http.response.start'],
    'status': int,
    'headers': Iterable[Tuple[bytes, bytes]],
    'trailers': bool
})

HTTPResponseBodyEvent = TypedDict('HTTPResponseBodyEvent', {
    'type': Literal['http.response.body'],
    'body': bytes,
    'more_body': bool
})

HTTPResponseTrailersEvent = TypedDict('HTTPResponseTrailersEvent', {
    'type': Literal['http.response.trailers'],
    'headers': Iterable[Tuple[bytes, bytes]],
    'more_trailers': bool
})

# HTTP Server Push (HTTP/2)
HTTPServerPushEvent = TypedDict('HTTPServerPushEvent', {
    'type': Literal['http.response.push'],
    'path': str,
    'headers': Iterable[Tuple[bytes, bytes]]
})

# HTTP Response Path Send
HTTPResponsePathsendEvent = TypedDict('HTTPResponsePathsendEvent', {
    'type': Literal['http.response.pathsend'],
    'path': str
})

WebSocket Event Types

Type definitions for WebSocket protocol events and connection management.

# WebSocket Connection Events
WebSocketConnectEvent = TypedDict('WebSocketConnectEvent', {
    'type': Literal['websocket.connect']
})

WebSocketAcceptEvent = TypedDict('WebSocketAcceptEvent', {
    'type': Literal['websocket.accept'],
    'subprotocol': Optional[str],
    'headers': Iterable[Tuple[bytes, bytes]]
})

# WebSocket Message Events
WebSocketReceiveEvent = TypedDict('WebSocketReceiveEvent', {
    'type': Literal['websocket.receive'],
    'bytes': Optional[bytes],
    'text': Optional[str]
})

WebSocketSendEvent = TypedDict('WebSocketSendEvent', {
    'type': Literal['websocket.send'],
    'bytes': Optional[bytes],
    'text': Optional[str]
})

# WebSocket Close Events
WebSocketDisconnectEvent = TypedDict('WebSocketDisconnectEvent', {
    'type': Literal['websocket.disconnect'],
    'code': int
})

WebSocketCloseEvent = TypedDict('WebSocketCloseEvent', {
    'type': Literal['websocket.close'],
    'code': int,
    'reason': Optional[str]
})

# WebSocket Response Events (for compatibility)
WebSocketResponseStartEvent = TypedDict('WebSocketResponseStartEvent', {
    'type': Literal['websocket.http.response.start'],
    'status': int,
    'headers': Iterable[Tuple[bytes, bytes]]
})

WebSocketResponseBodyEvent = TypedDict('WebSocketResponseBodyEvent', {
    'type': Literal['websocket.http.response.body'],
    'body': bytes,
    'more_body': bool
})

Lifespan Event Types

Type definitions for application lifespan management events.

# Lifespan Startup Events
LifespanStartupEvent = TypedDict('LifespanStartupEvent', {
    'type': Literal['lifespan.startup']
})

LifespanStartupCompleteEvent = TypedDict('LifespanStartupCompleteEvent', {
    'type': Literal['lifespan.startup.complete']
})

LifespanStartupFailedEvent = TypedDict('LifespanStartupFailedEvent', {
    'type': Literal['lifespan.startup.failed'],
    'message': str
})

# Lifespan Shutdown Events
LifespanShutdownEvent = TypedDict('LifespanShutdownEvent', {
    'type': Literal['lifespan.shutdown']
})

LifespanShutdownCompleteEvent = TypedDict('LifespanShutdownCompleteEvent', {
    'type': Literal['lifespan.shutdown.complete']
})

LifespanShutdownFailedEvent = TypedDict('LifespanShutdownFailedEvent', {
    'type': Literal['lifespan.shutdown.failed'],
    'message': str
})

Version and Utility Types

ASGI version specifications and utility type definitions.

# ASGI Version Types
ASGIVersions = TypedDict('ASGIVersions', {
    'spec_version': str,
    'version': Union[Literal['2.0'], Literal['3.0']]
})

# Common ASGI version values:
# {'version': '3.0', 'spec_version': '2.3'}
# {'version': '2.1', 'spec_version': '2.1'}

Usage Examples

Type-Safe ASGI Application

from asgiref.typing import ASGIApplication, HTTPScope, ASGIReceiveCallable, ASGISendCallable
from typing import cast

async def typed_http_app(scope: HTTPScope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
    """Type-safe HTTP ASGI application."""
    assert scope['type'] == 'http'
    
    # Type-safe access to scope properties
    method: str = scope['method']
    path: str = scope['path']
    headers: list[tuple[bytes, bytes]] = scope['headers']
    
    # Type-safe message handling
    request_message = await receive()
    if request_message['type'] == 'http.request':
        body: bytes = request_message['body']
        more_body: bool = request_message['more_body']
    
    # Type-safe response sending
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [[b'content-type', b'application/json']],
    })
    
    await send({
        'type': 'http.response.body',
        'body': b'{"message": "Type-safe response"}',
        'more_body': False,
    })

# Type annotation for the application
app: ASGIApplication = typed_http_app

Type-Safe WebSocket Application

from asgiref.typing import WebSocketScope, ASGIReceiveCallable, ASGISendCallable
import json

async def typed_websocket_app(scope: WebSocketScope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
    """Type-safe WebSocket ASGI application."""
    assert scope['type'] == 'websocket'
    
    # Type-safe scope access
    path: str = scope['path']
    subprotocols: list[str] = scope['subprotocols']
    
    # Accept connection
    await send({
        'type': 'websocket.accept',
        'subprotocol': subprotocols[0] if subprotocols else None,
    })
    
    # Message handling loop
    while True:
        message = await receive()
        
        if message['type'] == 'websocket.disconnect':
            code: int = message['code']
            print(f"WebSocket disconnected with code: {code}")
            break
            
        elif message['type'] == 'websocket.receive':
            # Type-safe message access
            text_data: str | None = message.get('text')
            bytes_data: bytes | None = message.get('bytes')
            
            if text_data:
                # Echo text message
                await send({
                    'type': 'websocket.send',
                    'text': f"Echo: {text_data}",
                })

Type-Safe Middleware

from asgiref.typing import ASGIApplication, Scope, ASGIReceiveCallable, ASGISendCallable
from typing import Callable

class TypedMiddleware:
    """Type-safe ASGI middleware."""
    
    def __init__(self, app: ASGIApplication) -> None:
        self.app = app
    
    async def __call__(self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
        """Type-safe middleware call."""
        if scope['type'] == 'http':
            # Add custom header to HTTP responses
            async def send_wrapper(message):
                if message['type'] == 'http.response.start':
                    headers = list(message.get('headers', []))
                    headers.append([b'x-middleware', b'typed'])
                    message = {**message, 'headers': headers}
                await send(message)
            
            await self.app(scope, receive, send_wrapper)
        else:
            # Pass through other protocols unchanged
            await self.app(scope, receive, send)

# Type-safe middleware factory
def create_typed_middleware(app: ASGIApplication) -> ASGIApplication:
    """Create type-safe middleware."""
    return TypedMiddleware(app)

Type-Safe Server Implementation

from asgiref.typing import ASGIApplication, HTTPScope, WebSocketScope, LifespanScope
from typing import Union

class TypedASGIServer:
    """Type-safe ASGI server implementation."""
    
    def __init__(self, app: ASGIApplication) -> None:
        self.app = app
    
    async def handle_http(self, scope: HTTPScope) -> None:
        """Handle HTTP requests with type safety."""
        async def receive():
            return {
                'type': 'http.request',
                'body': b'',
                'more_body': False,
            }
        
        async def send(message):
            if message['type'] == 'http.response.start':
                status: int = message['status']
                headers: list[tuple[bytes, bytes]] = message.get('headers', [])
                print(f"HTTP {status} with {len(headers)} headers")
            elif message['type'] == 'http.response.body':
                body: bytes = message.get('body', b'')
                print(f"HTTP body: {len(body)} bytes")
        
        await self.app(scope, receive, send)
    
    async def handle_websocket(self, scope: WebSocketScope) -> None:
        """Handle WebSocket connections with type safety."""
        async def receive():
            return {'type': 'websocket.connect'}
        
        async def send(message):
            if message['type'] == 'websocket.accept':
                subprotocol: str | None = message.get('subprotocol')
                print(f"WebSocket accepted with subprotocol: {subprotocol}")
        
        await self.app(scope, receive, send)
    
    async def handle_lifespan(self, scope: LifespanScope) -> None:
        """Handle lifespan events with type safety."""
        async def receive():
            return {'type': 'lifespan.startup'}
        
        async def send(message):
            if message['type'] == 'lifespan.startup.complete':
                print("Application startup complete")
        
        await self.app(scope, receive, send)

Type-Safe Testing

from asgiref.typing import ASGIApplication, HTTPScope
from asgiref.testing import ApplicationCommunicator

async def test_typed_application(app: ASGIApplication) -> None:
    """Type-safe application testing."""
    
    # Create type-safe scope
    scope: HTTPScope = {
        'type': 'http',
        'asgi': {'version': '3.0', 'spec_version': '2.3'},
        'http_version': '1.1',
        'method': 'GET',
        'scheme': 'http',
        'path': '/',
        'raw_path': b'/',
        'query_string': b'',
        'root_path': '',
        'headers': [[b'host', b'example.com']],
        'server': ('127.0.0.1', 8000),
        'client': ('127.0.0.1', 12345),
        'state': {},
        'extensions': {},
    }
    
    # Type-safe testing
    communicator = ApplicationCommunicator(app, scope)
    
    try:
        # Send typed request
        await communicator.send_input({
            'type': 'http.request',
            'body': b'test data',
            'more_body': False,
        })
        
        # Receive typed response
        response_start = await communicator.receive_output()
        assert response_start['type'] == 'http.response.start'
        
        status: int = response_start['status']
        headers: list[tuple[bytes, bytes]] = response_start.get('headers', [])
        
        print(f"Response: {status} with {len(headers)} headers")
        
    finally:
        await communicator.stop()

Protocol Detection Utility

from asgiref.typing import Scope, HTTPScope, WebSocketScope, LifespanScope
from typing import TypeGuard

def is_http_scope(scope: Scope) -> TypeGuard[HTTPScope]:
    """Type guard for HTTP scopes."""
    return scope['type'] == 'http'

def is_websocket_scope(scope: Scope) -> TypeGuard[WebSocketScope]:
    """Type guard for WebSocket scopes."""
    return scope['type'] == 'websocket'

def is_lifespan_scope(scope: Scope) -> TypeGuard[LifespanScope]:
    """Type guard for Lifespan scopes."""
    return scope['type'] == 'lifespan'

async def protocol_aware_app(scope: Scope, receive, send) -> None:
    """Application that uses type guards for protocol detection."""
    
    if is_http_scope(scope):
        # TypeScript-style type narrowing
        method: str = scope['method']  # Type checker knows this is HTTPScope
        path: str = scope['path']
        print(f"HTTP {method} {path}")
        
    elif is_websocket_scope(scope):
        # Type checker knows this is WebSocketScope
        subprotocols: list[str] = scope['subprotocols']
        print(f"WebSocket with subprotocols: {subprotocols}")
        
    elif is_lifespan_scope(scope):
        # Type checker knows this is LifespanScope
        print("Lifespan event")

Key Type Safety Features

The type definitions provide:

  • Complete Protocol Coverage: All ASGI 2 and ASGI 3 events and scopes
  • TypedDict Support: Structured dictionary types with required/optional fields
  • Union Types: Proper type unions for events and scopes
  • Generic Application Types: Support for both ASGI2 and ASGI3 applications
  • Type Guards: Utilities for runtime type checking and narrowing
  • IDE Integration: Full autocomplete and type checking in modern Python IDEs

Install with Tessl CLI

npx tessl i tessl/pypi-asgiref

docs

compatibility.md

current-thread-executor.md

index.md

local-storage.md

server-base.md

sync-async.md

testing.md

timeout.md

type-definitions.md

wsgi-integration.md

tile.json