0
# ASGI Compatibility
1

2
Utilities for ensuring ASGI applications work consistently across different ASGI versions and server implementations. These functions handle the transition from ASGI2 (double-callable) to ASGI3 (single-callable) patterns automatically.
3

4
## Capabilities
5

6
### Application Style Detection
7

8
Detect whether an ASGI application uses the legacy double-callable pattern or the modern single-callable pattern.
9

10
```python { .api }
11
def is_double_callable(application):
12
    """
13
    Test if an application is legacy-style (double-callable).
14
    
15
    Parameters:
16
    - application: callable, ASGI application to test
17
    
18
    Returns:
19
    bool: True if application uses double-callable pattern (ASGI2), False otherwise
20
    """
21
```
22

23
### Application Style Conversion
24

25
Convert ASGI applications between different callable patterns to ensure compatibility with all ASGI servers.
26

27
```python { .api }
28
def double_to_single_callable(application):
29
    """
30
    Transform double-callable ASGI app to single-callable.
31
    
32
    Parameters:
33
    - application: callable, double-callable ASGI2 application
34
    
35
    Returns:
36
    callable: Single-callable ASGI3 application
37
    """
38

39
def guarantee_single_callable(application):
40
    """
41
    Ensure application is in single-callable ASGI3 style.
42
    
43
    Parameters:
44
    - application: callable, ASGI application of any style
45
    
46
    Returns:
47
    callable: Single-callable ASGI3 application (converted if necessary)
48
    """
49
```
50

51
## Usage Examples
52

53
### Basic Compatibility Checking
54

55
```python
56
from asgiref.compatibility import is_double_callable, guarantee_single_callable
57

58
# ASGI2-style application (legacy double-callable)
59
def asgi2_app(scope):
60
    async def asgi_coroutine(receive, send):
61
        await send({
62
            'type': 'http.response.start',
63
            'status': 200,
64
            'headers': [[b'content-type', b'text/plain']],
65
        })
66
        await send({
67
            'type': 'http.response.body',
68
            'body': b'Hello from ASGI2',
69
        })
70
    return asgi_coroutine
71

72
# ASGI3-style application (modern single-callable)
73
async def asgi3_app(scope, receive, send):
74
    await send({
75
        'type': 'http.response.start',
76
        'status': 200,
77
        'headers': [[b'content-type', b'text/plain']],
78
    })
79
    await send({
80
        'type': 'http.response.body',
81
        'body': b'Hello from ASGI3',
82
    })
83

84
# Check application styles
85
print(is_double_callable(asgi2_app))  # True
86
print(is_double_callable(asgi3_app))  # False
87

88
# Ensure both are ASGI3 compatible
89
compatible_app1 = guarantee_single_callable(asgi2_app)
90
compatible_app2 = guarantee_single_callable(asgi3_app)
91

92
print(is_double_callable(compatible_app1))  # False
93
print(is_double_callable(compatible_app2))  # False
94
```
95

96
### Framework Integration
97

98
```python
99
from asgiref.compatibility import guarantee_single_callable
100

101
class ASGIFramework:
102
    def __init__(self):
103
        self.routes = []
104
    
105
    def add_app(self, path, app):
106
        """Add an application, ensuring ASGI3 compatibility."""
107
        compatible_app = guarantee_single_callable(app)
108
        self.routes.append((path, compatible_app))
109
    
110
    async def __call__(self, scope, receive, send):
111
        path = scope['path']
112
        
113
        for route_path, app in self.routes:
114
            if path.startswith(route_path):
115
                await app(scope, receive, send)
116
                return
117
        
118
        # 404 response
119
        await send({
120
            'type': 'http.response.start',
121
            'status': 404,
122
            'headers': [[b'content-type', b'text/plain']],
123
        })
124
        await send({
125
            'type': 'http.response.body',
126
            'body': b'Not Found',
127
        })
128

129
# Usage
130
framework = ASGIFramework()
131

132
# Can add both ASGI2 and ASGI3 apps
133
framework.add_app('/legacy', asgi2_app)
134
framework.add_app('/modern', asgi3_app)
135
```
136

137
### Middleware Compatibility
138

139
```python
140
from asgiref.compatibility import guarantee_single_callable
141

142
class CompatibilityMiddleware:
143
    """Middleware that ensures all wrapped applications are ASGI3 compatible."""
144
    
145
    def __init__(self, app):
146
        self.app = guarantee_single_callable(app)
147
    
148
    async def __call__(self, scope, receive, send):
149
        # Add compatibility headers
150
        async def send_wrapper(message):
151
            if message['type'] == 'http.response.start':
152
                headers = list(message.get('headers', []))
153
                headers.append([b'x-asgi-version', b'3.0'])
154
                message = {**message, 'headers': headers}
155
            await send(message)
156
        
157
        await self.app(scope, receive, send_wrapper)
158

159
# Wrap any ASGI application with compatibility middleware
160
def create_compatible_app(app):
161
    return CompatibilityMiddleware(app)
162

163
# Works with both ASGI2 and ASGI3 applications
164
compatible_legacy = create_compatible_app(asgi2_app)
165
compatible_modern = create_compatible_app(asgi3_app)
166
```
167

168
### Server Implementation
169

170
```python
171
from asgiref.compatibility import guarantee_single_callable, is_double_callable
172
import asyncio
173

174
class SimpleASGIServer:
175
    def __init__(self, app, host='127.0.0.1', port=8000):
176
        self.app = guarantee_single_callable(app)
177
        self.host = host
178
        self.port = port
179
    
180
    async def handle_request(self, reader, writer):
181
        """Handle a single HTTP request."""
182
        # Read request (simplified)
183
        request_line = await reader.readline()
184
        
185
        # Create ASGI scope
186
        scope = {
187
            'type': 'http',
188
            'method': 'GET',
189
            'path': '/',
190
            'headers': [],
191
            'query_string': b'',
192
        }
193
        
194
        # ASGI receive callable
195
        async def receive():
196
            return {'type': 'http.request', 'body': b''}
197
        
198
        # ASGI send callable
199
        async def send(message):
200
            if message['type'] == 'http.response.start':
201
                status = message['status']
202
                writer.write(f'HTTP/1.1 {status} OK\\r\\n'.encode())
203
                for name, value in message.get('headers', []):
204
                    writer.write(f'{name.decode()}: {value.decode()}\\r\\n'.encode())
205
                writer.write(b'\\r\\n')
206
            elif message['type'] == 'http.response.body':
207
                body = message.get('body', b'')
208
                writer.write(body)
209
                writer.close()
210
        
211
        # Call ASGI application
212
        await self.app(scope, receive, send)
213
    
214
    def run(self):
215
        """Run the server."""
216
        print(f"Starting server on {self.host}:{self.port}")
217
        
218
        # Detect application type for logging
219
        app_type = "ASGI2 (converted)" if is_double_callable(self.app) else "ASGI3"
220
        print(f"Application type: {app_type}")
221
        
222
        # Start server (simplified)
223
        loop = asyncio.get_event_loop()
224
        server = loop.run_until_complete(
225
            asyncio.start_server(self.handle_request, self.host, self.port)
226
        )
227
        loop.run_until_complete(server.serve_forever())
228

229
# Can serve any ASGI application
230
server = SimpleASGIServer(asgi2_app)  # Automatically converted to ASGI3
231
# server.run()
232
```