0
# Type Definitions
1

2
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.
3

4
## Capabilities
5

6
### Protocol Types
7

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

10
```python { .api }
11
# Application Types
12
ASGIApplication = Union[ASGI2Application, ASGI3Application]
13
ASGI2Application = Callable[[Scope], ASGI2Protocol]
14
ASGI3Application = Callable[[Scope, ASGIReceiveCallable, ASGISendCallable], Awaitable[None]]
15
ASGI2Protocol = Callable[[ASGIReceiveCallable, ASGISendCallable], Awaitable[None]]
16

17
# Channel Types  
18
ASGIReceiveCallable = Callable[[], Awaitable[ASGIReceiveEvent]]
19
ASGISendCallable = Callable[[ASGISendEvent], Awaitable[None]]
20

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

26
### Scope Types
27

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

30
```python { .api }
31
# Main Scope Union
32
Scope = Union[HTTPScope, WebSocketScope, LifespanScope]
33

34
# HTTP Scope
35
HTTPScope = TypedDict('HTTPScope', {
36
    'type': Literal['http'],
37
    'asgi': ASGIVersions,
38
    'http_version': str,
39
    'method': str,
40
    'scheme': str,
41
    'path': str,
42
    'raw_path': bytes,
43
    'query_string': bytes,
44
    'root_path': str,
45
    'headers': Iterable[Tuple[bytes, bytes]],
46
    'client': Optional[Tuple[str, int]],
47
    'server': Optional[Tuple[str, Optional[int]]],
48
    'state': NotRequired[Dict[str, Any]],
49
    'extensions': Optional[Dict[str, Dict[object, object]]]
50
})
51

52
# WebSocket Scope  
53
WebSocketScope = TypedDict('WebSocketScope', {
54
    'type': Literal['websocket'],
55
    'asgi': ASGIVersions,
56
    'http_version': str,
57
    'scheme': str,
58
    'path': str,
59
    'raw_path': bytes,
60
    'query_string': bytes,
61
    'root_path': str,
62
    'headers': Iterable[Tuple[bytes, bytes]],
63
    'client': Optional[Tuple[str, int]],
64
    'server': Optional[Tuple[str, Optional[int]]],
65
    'subprotocols': Iterable[str],
66
    'state': NotRequired[Dict[str, Any]],
67
    'extensions': Optional[Dict[str, Dict[object, object]]]
68
})
69

70
# Lifespan Scope
71
LifespanScope = TypedDict('LifespanScope', {
72
    'type': Literal['lifespan'],
73
    'asgi': ASGIVersions,
74
    'state': NotRequired[Dict[str, Any]],
75
    'extensions': Optional[Dict[str, Dict[object, object]]]
76
})
77

78
# Legacy WWW Scope (deprecated)
79
WWWScope = Union[HTTPScope, WebSocketScope]
80
```
81

82
### HTTP Event Types
83

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

86
```python { .api }
87
# HTTP Request Events
88
HTTPRequestEvent = TypedDict('HTTPRequestEvent', {
89
    'type': Literal['http.request'],
90
    'body': bytes,
91
    'more_body': bool
92
})
93

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

98
# HTTP Response Events
99
HTTPResponseStartEvent = TypedDict('HTTPResponseStartEvent', {
100
    'type': Literal['http.response.start'],
101
    'status': int,
102
    'headers': Iterable[Tuple[bytes, bytes]],
103
    'trailers': bool
104
})
105

106
HTTPResponseBodyEvent = TypedDict('HTTPResponseBodyEvent', {
107
    'type': Literal['http.response.body'],
108
    'body': bytes,
109
    'more_body': bool
110
})
111

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

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

125
# HTTP Response Path Send
126
HTTPResponsePathsendEvent = TypedDict('HTTPResponsePathsendEvent', {
127
    'type': Literal['http.response.pathsend'],
128
    'path': str
129
})
130
```
131

132
### WebSocket Event Types
133

134
Type definitions for WebSocket protocol events and connection management.
135

136
```python { .api }
137
# WebSocket Connection Events
138
WebSocketConnectEvent = TypedDict('WebSocketConnectEvent', {
139
    'type': Literal['websocket.connect']
140
})
141

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

148
# WebSocket Message Events
149
WebSocketReceiveEvent = TypedDict('WebSocketReceiveEvent', {
150
    'type': Literal['websocket.receive'],
151
    'bytes': Optional[bytes],
152
    'text': Optional[str]
153
})
154

155
WebSocketSendEvent = TypedDict('WebSocketSendEvent', {
156
    'type': Literal['websocket.send'],
157
    'bytes': Optional[bytes],
158
    'text': Optional[str]
159
})
160

161
# WebSocket Close Events
162
WebSocketDisconnectEvent = TypedDict('WebSocketDisconnectEvent', {
163
    'type': Literal['websocket.disconnect'],
164
    'code': int
165
})
166

167
WebSocketCloseEvent = TypedDict('WebSocketCloseEvent', {
168
    'type': Literal['websocket.close'],
169
    'code': int,
170
    'reason': Optional[str]
171
})
172

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

180
WebSocketResponseBodyEvent = TypedDict('WebSocketResponseBodyEvent', {
181
    'type': Literal['websocket.http.response.body'],
182
    'body': bytes,
183
    'more_body': bool
184
})
185
```
186

187
### Lifespan Event Types
188

189
Type definitions for application lifespan management events.
190

191
```python { .api }
192
# Lifespan Startup Events
193
LifespanStartupEvent = TypedDict('LifespanStartupEvent', {
194
    'type': Literal['lifespan.startup']
195
})
196

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

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

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

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

215
LifespanShutdownFailedEvent = TypedDict('LifespanShutdownFailedEvent', {
216
    'type': Literal['lifespan.shutdown.failed'],
217
    'message': str
218
})
219
```
220

221
### Version and Utility Types
222

223
ASGI version specifications and utility type definitions.
224

225
```python { .api }
226
# ASGI Version Types
227
ASGIVersions = TypedDict('ASGIVersions', {
228
    'spec_version': str,
229
    'version': Union[Literal['2.0'], Literal['3.0']]
230
})
231

232
# Common ASGI version values:
233
# {'version': '3.0', 'spec_version': '2.3'}
234
# {'version': '2.1', 'spec_version': '2.1'}
235
```
236

237
## Usage Examples
238

239
### Type-Safe ASGI Application
240

241
```python
242
from asgiref.typing import ASGIApplication, HTTPScope, ASGIReceiveCallable, ASGISendCallable
243
from typing import cast
244

245
async def typed_http_app(scope: HTTPScope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
246
    """Type-safe HTTP ASGI application."""
247
    assert scope['type'] == 'http'
248
    
249
    # Type-safe access to scope properties
250
    method: str = scope['method']
251
    path: str = scope['path']
252
    headers: list[tuple[bytes, bytes]] = scope['headers']
253
    
254
    # Type-safe message handling
255
    request_message = await receive()
256
    if request_message['type'] == 'http.request':
257
        body: bytes = request_message['body']
258
        more_body: bool = request_message['more_body']
259
    
260
    # Type-safe response sending
261
    await send({
262
        'type': 'http.response.start',
263
        'status': 200,
264
        'headers': [[b'content-type', b'application/json']],
265
    })
266
    
267
    await send({
268
        'type': 'http.response.body',
269
        'body': b'{"message": "Type-safe response"}',
270
        'more_body': False,
271
    })
272

273
# Type annotation for the application
274
app: ASGIApplication = typed_http_app
275
```
276

277
### Type-Safe WebSocket Application
278

279
```python
280
from asgiref.typing import WebSocketScope, ASGIReceiveCallable, ASGISendCallable
281
import json
282

283
async def typed_websocket_app(scope: WebSocketScope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
284
    """Type-safe WebSocket ASGI application."""
285
    assert scope['type'] == 'websocket'
286
    
287
    # Type-safe scope access
288
    path: str = scope['path']
289
    subprotocols: list[str] = scope['subprotocols']
290
    
291
    # Accept connection
292
    await send({
293
        'type': 'websocket.accept',
294
        'subprotocol': subprotocols[0] if subprotocols else None,
295
    })
296
    
297
    # Message handling loop
298
    while True:
299
        message = await receive()
300
        
301
        if message['type'] == 'websocket.disconnect':
302
            code: int = message['code']
303
            print(f"WebSocket disconnected with code: {code}")
304
            break
305
            
306
        elif message['type'] == 'websocket.receive':
307
            # Type-safe message access
308
            text_data: str | None = message.get('text')
309
            bytes_data: bytes | None = message.get('bytes')
310
            
311
            if text_data:
312
                # Echo text message
313
                await send({
314
                    'type': 'websocket.send',
315
                    'text': f"Echo: {text_data}",
316
                })
317
```
318

319
### Type-Safe Middleware
320

321
```python
322
from asgiref.typing import ASGIApplication, Scope, ASGIReceiveCallable, ASGISendCallable
323
from typing import Callable
324

325
class TypedMiddleware:
326
    """Type-safe ASGI middleware."""
327
    
328
    def __init__(self, app: ASGIApplication) -> None:
329
        self.app = app
330
    
331
    async def __call__(self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
332
        """Type-safe middleware call."""
333
        if scope['type'] == 'http':
334
            # Add custom header to HTTP responses
335
            async def send_wrapper(message):
336
                if message['type'] == 'http.response.start':
337
                    headers = list(message.get('headers', []))
338
                    headers.append([b'x-middleware', b'typed'])
339
                    message = {**message, 'headers': headers}
340
                await send(message)
341
            
342
            await self.app(scope, receive, send_wrapper)
343
        else:
344
            # Pass through other protocols unchanged
345
            await self.app(scope, receive, send)
346

347
# Type-safe middleware factory
348
def create_typed_middleware(app: ASGIApplication) -> ASGIApplication:
349
    """Create type-safe middleware."""
350
    return TypedMiddleware(app)
351
```
352

353
### Type-Safe Server Implementation
354

355
```python
356
from asgiref.typing import ASGIApplication, HTTPScope, WebSocketScope, LifespanScope
357
from typing import Union
358

359
class TypedASGIServer:
360
    """Type-safe ASGI server implementation."""
361
    
362
    def __init__(self, app: ASGIApplication) -> None:
363
        self.app = app
364
    
365
    async def handle_http(self, scope: HTTPScope) -> None:
366
        """Handle HTTP requests with type safety."""
367
        async def receive():
368
            return {
369
                'type': 'http.request',
370
                'body': b'',
371
                'more_body': False,
372
            }
373
        
374
        async def send(message):
375
            if message['type'] == 'http.response.start':
376
                status: int = message['status']
377
                headers: list[tuple[bytes, bytes]] = message.get('headers', [])
378
                print(f"HTTP {status} with {len(headers)} headers")
379
            elif message['type'] == 'http.response.body':
380
                body: bytes = message.get('body', b'')
381
                print(f"HTTP body: {len(body)} bytes")
382
        
383
        await self.app(scope, receive, send)
384
    
385
    async def handle_websocket(self, scope: WebSocketScope) -> None:
386
        """Handle WebSocket connections with type safety."""
387
        async def receive():
388
            return {'type': 'websocket.connect'}
389
        
390
        async def send(message):
391
            if message['type'] == 'websocket.accept':
392
                subprotocol: str | None = message.get('subprotocol')
393
                print(f"WebSocket accepted with subprotocol: {subprotocol}")
394
        
395
        await self.app(scope, receive, send)
396
    
397
    async def handle_lifespan(self, scope: LifespanScope) -> None:
398
        """Handle lifespan events with type safety."""
399
        async def receive():
400
            return {'type': 'lifespan.startup'}
401
        
402
        async def send(message):
403
            if message['type'] == 'lifespan.startup.complete':
404
                print("Application startup complete")
405
        
406
        await self.app(scope, receive, send)
407
```
408

409
### Type-Safe Testing
410

411
```python
412
from asgiref.typing import ASGIApplication, HTTPScope
413
from asgiref.testing import ApplicationCommunicator
414

415
async def test_typed_application(app: ASGIApplication) -> None:
416
    """Type-safe application testing."""
417
    
418
    # Create type-safe scope
419
    scope: HTTPScope = {
420
        'type': 'http',
421
        'asgi': {'version': '3.0', 'spec_version': '2.3'},
422
        'http_version': '1.1',
423
        'method': 'GET',
424
        'scheme': 'http',
425
        'path': '/',
426
        'raw_path': b'/',
427
        'query_string': b'',
428
        'root_path': '',
429
        'headers': [[b'host', b'example.com']],
430
        'server': ('127.0.0.1', 8000),
431
        'client': ('127.0.0.1', 12345),
432
        'state': {},
433
        'extensions': {},
434
    }
435
    
436
    # Type-safe testing
437
    communicator = ApplicationCommunicator(app, scope)
438
    
439
    try:
440
        # Send typed request
441
        await communicator.send_input({
442
            'type': 'http.request',
443
            'body': b'test data',
444
            'more_body': False,
445
        })
446
        
447
        # Receive typed response
448
        response_start = await communicator.receive_output()
449
        assert response_start['type'] == 'http.response.start'
450
        
451
        status: int = response_start['status']
452
        headers: list[tuple[bytes, bytes]] = response_start.get('headers', [])
453
        
454
        print(f"Response: {status} with {len(headers)} headers")
455
        
456
    finally:
457
        await communicator.stop()
458
```
459

460
### Protocol Detection Utility
461

462
```python
463
from asgiref.typing import Scope, HTTPScope, WebSocketScope, LifespanScope
464
from typing import TypeGuard
465

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

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

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

478
async def protocol_aware_app(scope: Scope, receive, send) -> None:
479
    """Application that uses type guards for protocol detection."""
480
    
481
    if is_http_scope(scope):
482
        # TypeScript-style type narrowing
483
        method: str = scope['method']  # Type checker knows this is HTTPScope
484
        path: str = scope['path']
485
        print(f"HTTP {method} {path}")
486
        
487
    elif is_websocket_scope(scope):
488
        # Type checker knows this is WebSocketScope
489
        subprotocols: list[str] = scope['subprotocols']
490
        print(f"WebSocket with subprotocols: {subprotocols}")
491
        
492
    elif is_lifespan_scope(scope):
493
        # Type checker knows this is LifespanScope
494
        print("Lifespan event")
495
```
496

497
## Key Type Safety Features
498

499
The type definitions provide:
500

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