0
# Web Framework Integration
1

2
Middleware classes and utilities for integrating Socket.IO servers with popular Python web frameworks. These integration layers enable Socket.IO to work seamlessly with WSGI and ASGI applications while providing static file serving and flexible routing options.
3

4
## Capabilities
5

6
### WSGIApp
7

8
WSGI middleware that integrates a Socket.IO server with WSGI-compatible web frameworks like Flask, Django, and others.
9

10
```python { .api }
11
class WSGIApp:
12
    """
13
    WSGI middleware for Socket.IO applications.
14
    
15
    Inherits from: engineio.WSGIApp
16
    
17
    Attributes:
18
        socketio_app: Socket.IO server instance
19
        wsgi_app: Wrapped WSGI application
20
        socketio_path (str): Socket.IO endpoint path
21
    """
22
    
23
    def __init__(self, socketio_app, wsgi_app=None, static_files=None, socketio_path='socket.io'):
24
        """
25
        Initialize the WSGI middleware.
26
        
27
        Args:
28
            socketio_app: Socket.IO server instance
29
            wsgi_app: WSGI application to wrap (optional)
30
            static_files (dict): Static file mappings {'/path': 'directory'}
31
            socketio_path (str): Socket.IO endpoint path (default: 'socket.io')
32
        """
33
    
34
    def __call__(self, environ, start_response):
35
        """
36
        WSGI application callable.
37
        
38
        Args:
39
            environ (dict): WSGI environment
40
            start_response (callable): WSGI start_response function
41
            
42
        Returns:
43
            WSGI response iterable
44
        """
45
```
46

47
#### Usage Examples
48

49
##### Standalone Socket.IO Application
50

51
```python
52
import socketio
53

54
# Create Socket.IO server
55
sio = socketio.Server(cors_allowed_origins="*")
56

57
@sio.event
58
def connect(sid, environ):
59
    print(f'Client {sid} connected')
60

61
@sio.event
62
def disconnect(sid):
63
    print(f'Client {sid} disconnected')
64

65
@sio.event
66
def message(sid, data):
67
    sio.emit('response', {'echo': data})
68

69
# Create WSGI app with static files
70
app = socketio.WSGIApp(sio, static_files={
71
    '/': 'static/',
72
    '/css': 'static/css/',
73
    '/js': 'static/js/'
74
})
75

76
if __name__ == '__main__':
77
    import eventlet
78
    import eventlet.wsgi
79
    eventlet.wsgi.server(eventlet.listen(('', 5000)), app)
80
```
81

82
##### Flask Integration
83

84
```python
85
import socketio
86
from flask import Flask
87

88
# Create Flask app
89
flask_app = Flask(__name__)
90

91
# Create Socket.IO server
92
sio = socketio.Server(cors_allowed_origins="*")
93

94
@flask_app.route('/')
95
def index():
96
    return '<h1>Flask + Socket.IO</h1>'
97

98
@flask_app.route('/api/data')
99
def get_data():
100
    return {'message': 'Hello from Flask!'}
101

102
@sio.event
103
def connect(sid, environ):
104
    print(f'Client {sid} connected')
105

106
@sio.event
107
def flask_message(sid, data):
108
    # Can access Flask app context if needed
109
    with flask_app.app_context():
110
        # Flask operations here
111
        pass
112
    sio.emit('flask_response', {'processed': data})
113

114
# Wrap Flask app with Socket.IO
115
app = socketio.WSGIApp(sio, flask_app)
116

117
if __name__ == '__main__':
118
    import eventlet
119
    import eventlet.wsgi
120
    eventlet.wsgi.server(eventlet.listen(('', 5000)), app)
121
```
122

123
##### Django Integration
124

125
```python
126
import socketio
127
import os
128
from django.core.wsgi import get_wsgi_application
129

130
# Django setup
131
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
132
django_app = get_wsgi_application()
133

134
# Create Socket.IO server
135
sio = socketio.Server(cors_allowed_origins="*")
136

137
@sio.event
138
def connect(sid, environ):
139
    print(f'Client {sid} connected')
140

141
@sio.event
142
def django_message(sid, data):
143
    # Can import and use Django models
144
    from myapp.models import Message
145
    message = Message.objects.create(content=data['text'])
146
    sio.emit('message_saved', {'id': message.id})
147

148
# Combine Django + Socket.IO
149
app = socketio.WSGIApp(sio, django_app)
150
```
151

152
##### Custom Routing
153

154
```python
155
import socketio
156

157
sio = socketio.Server()
158

159
@sio.event
160
def connect(sid, environ):
161
    print(f'Client connected: {sid}')
162

163
# Custom Socket.IO path
164
app = socketio.WSGIApp(sio, socketio_path='custom-socketio')
165

166
# Client would connect to: http://localhost:5000/custom-socketio/
167
```
168

169
### ASGIApp
170

171
ASGI middleware that integrates an AsyncServer with ASGI-compatible frameworks like FastAPI, Starlette, and Django Channels.
172

173
```python { .api }
174
class ASGIApp:
175
    """
176
    ASGI middleware for async Socket.IO applications.
177
    
178
    Inherits from: engineio.ASGIApp
179
    
180
    Attributes:
181
        socketio_server: AsyncServer instance
182
        other_asgi_app: Wrapped ASGI application
183
        socketio_path (str): Socket.IO endpoint path
184
    """
185
    
186
    def __init__(self, socketio_server, other_asgi_app=None, static_files=None, 
187
                 socketio_path='socket.io', on_startup=None, on_shutdown=None):
188
        """
189
        Initialize the ASGI middleware.
190
        
191
        Args:
192
            socketio_server: AsyncServer instance
193
            other_asgi_app: ASGI application to wrap (optional)
194
            static_files (dict): Static file mappings {'/path': 'directory'}
195
            socketio_path (str): Socket.IO endpoint path (default: 'socket.io')
196
            on_startup (list): List of startup event handlers
197
            on_shutdown (list): List of shutdown event handlers
198
        """
199
    
200
    async def __call__(self, scope, receive, send):
201
        """
202
        ASGI application callable.
203
        
204
        Args:
205
            scope (dict): ASGI connection scope
206
            receive (callable): ASGI receive function
207
            send (callable): ASGI send function
208
        """
209
```
210

211
#### Usage Examples
212

213
##### Standalone Async Socket.IO Application
214

215
```python
216
import socketio
217

218
# Create async Socket.IO server
219
sio = socketio.AsyncServer(cors_allowed_origins="*")
220

221
@sio.event
222
async def connect(sid, environ):
223
    print(f'Client {sid} connected')
224

225
@sio.event
226
async def disconnect(sid):
227
    print(f'Client {sid} disconnected')
228

229
@sio.event
230
async def async_message(sid, data):
231
    # Simulate async processing
232
    await asyncio.sleep(0.1)
233
    await sio.emit('async_response', {'processed': data})
234

235
# Create ASGI app
236
app = socketio.ASGIApp(sio, static_files={
237
    '/': 'static/',
238
})
239

240
# Run with: uvicorn main:app --host 0.0.0.0 --port 5000
241
```
242

243
##### FastAPI Integration
244

245
```python
246
import socketio
247
from fastapi import FastAPI
248

249
# Create FastAPI app
250
fastapi_app = FastAPI()
251

252
# Create async Socket.IO server
253
sio = socketio.AsyncServer(cors_allowed_origins="*")
254

255
@fastapi_app.get("/")
256
async def read_root():
257
    return {"message": "FastAPI + Socket.IO"}
258

259
@fastapi_app.get("/api/users/{user_id}")
260
async def get_user(user_id: int):
261
    return {"user_id": user_id, "name": f"User {user_id}"}
262

263
@sio.event
264
async def connect(sid, environ):
265
    print(f'Client {sid} connected')
266

267
@sio.event
268
async def fastapi_message(sid, data):
269
    # Can use FastAPI dependencies and features
270
    await sio.emit('fastapi_response', {'echo': data})
271

272
# Combine FastAPI + Socket.IO
273
app = socketio.ASGIApp(sio, fastapi_app)
274

275
# Run with: uvicorn main:app --host 0.0.0.0 --port 5000
276
```
277

278
##### Starlette Integration
279

280
```python
281
import socketio
282
from starlette.applications import Starlette
283
from starlette.responses import JSONResponse
284
from starlette.routing import Route
285

286
# Create async Socket.IO server
287
sio = socketio.AsyncServer(cors_allowed_origins="*")
288

289
async def homepage(request):
290
    return JSONResponse({'message': 'Starlette + Socket.IO'})
291

292
async def api_endpoint(request):
293
    return JSONResponse({'data': 'API response'})
294

295
# Create Starlette app
296
starlette_app = Starlette(routes=[
297
    Route('/', homepage),
298
    Route('/api/data', api_endpoint),
299
])
300

301
@sio.event
302
async def connect(sid, environ):
303
    print(f'Client {sid} connected')
304

305
@sio.event
306
async def starlette_message(sid, data):
307
    await sio.emit('starlette_response', {'received': data})
308

309
# Combine Starlette + Socket.IO
310
app = socketio.ASGIApp(sio, starlette_app)
311
```
312

313
##### Startup and Shutdown Handlers
314

315
```python
316
import socketio
317
from fastapi import FastAPI
318

319
sio = socketio.AsyncServer()
320
fastapi_app = FastAPI()
321

322
# Startup and shutdown handlers
323
async def startup_handler():
324
    print("Application starting up...")
325
    # Initialize resources, database connections, etc.
326

327
async def shutdown_handler():
328
    print("Application shutting down...")
329
    # Clean up resources
330

331
app = socketio.ASGIApp(
332
    sio, 
333
    fastapi_app,
334
    on_startup=[startup_handler],
335
    on_shutdown=[shutdown_handler]
336
)
337
```
338

339
### get_tornado_handler
340

341
Utility function to create a Tornado request handler for Socket.IO server integration.
342

343
```python { .api }
344
def get_tornado_handler(socketio_server):
345
    """
346
    Get Tornado request handler for Socket.IO server.
347
    
348
    Args:
349
        socketio_server: Socket.IO server instance (Server or AsyncServer)
350
        
351
    Returns:
352
        Tornado handler class for Socket.IO integration
353
    """
354
```
355

356
#### Usage Example
357

358
```python
359
import socketio
360
import tornado.web
361
import tornado.ioloop
362

363
# Create Socket.IO server
364
sio = socketio.Server()
365

366
@sio.event
367
def connect(sid, environ):
368
    print(f'Client {sid} connected')
369

370
@sio.event
371
def tornado_message(sid, data):
372
    sio.emit('tornado_response', {'echo': data})
373

374
# Create Tornado application
375
class MainHandler(tornado.web.RequestHandler):
376
    def get(self):
377
        self.write("Tornado + Socket.IO")
378

379
# Get Socket.IO handler for Tornado
380
SocketIOHandler = socketio.get_tornado_handler(sio)
381

382
app = tornado.web.Application([
383
    (r"/", MainHandler),
384
    (r"/socket.io/.*", SocketIOHandler),
385
], debug=True)
386

387
if __name__ == "__main__":
388
    app.listen(5000)
389
    print("Server running on http://localhost:5000")
390
    tornado.ioloop.IOLoop.current().start()
391
```
392

393
### Middleware (Deprecated)
394

395
```python { .api }
396
class Middleware:
397
    """
398
    Deprecated alias for WSGIApp.
399
    
400
    Note:
401
        This class is deprecated. Use WSGIApp instead.
402
    """
403
```
404

405
The Middleware class is a deprecated alias for WSGIApp. Use WSGIApp directly for new applications.
406

407
## Static File Serving
408

409
Both WSGIApp and ASGIApp support static file serving for client-side assets:
410

411
```python
412
# Static file mappings
413
static_files = {
414
    '/': 'public/',           # Serve public/ directory at root
415
    '/css': 'assets/css/',    # Serve CSS from assets/css/
416
    '/js': 'assets/js/',      # Serve JavaScript from assets/js/
417
    '/images': 'assets/img/'  # Serve images from assets/img/
418
}
419

420
# WSGI with static files
421
wsgi_app = socketio.WSGIApp(sio, flask_app, static_files=static_files)
422

423
# ASGI with static files
424
asgi_app = socketio.ASGIApp(sio, fastapi_app, static_files=static_files)
425
```
426

427
## CORS Configuration
428

429
Configure CORS (Cross-Origin Resource Sharing) for web browser clients:
430

431
```python
432
# Allow all origins (development only)
433
sio = socketio.Server(cors_allowed_origins="*")
434

435
# Allow specific origins
436
sio = socketio.Server(cors_allowed_origins=[
437
    "http://localhost:3000",
438
    "https://myapp.com"
439
])
440

441
# Async server CORS
442
async_sio = socketio.AsyncServer(cors_allowed_origins=[
443
    "http://localhost:8080",
444
    "https://app.example.com"
445
])
446
```
447

448
## Custom Socket.IO Endpoints
449

450
Change the default Socket.IO endpoint path:
451

452
```python
453
# Custom endpoint path
454
app = socketio.WSGIApp(sio, socketio_path='realtime')
455
# Clients connect to: /realtime/
456

457
app = socketio.ASGIApp(sio, socketio_path='ws')
458
# Clients connect to: /ws/
459
```
460

461
## Production Deployment
462

463
### WSGI Deployment
464

465
```python
466
# With Gunicorn + eventlet
467
# gunicorn --worker-class eventlet -w 1 main:app
468

469
# With uWSGI + gevent
470
# uwsgi --http :5000 --gevent 1000 --module main:app
471
```
472

473
### ASGI Deployment
474

475
```python
476
# With Uvicorn
477
# uvicorn main:app --host 0.0.0.0 --port 5000
478

479
# With Hypercorn
480
# hypercorn main:app --bind 0.0.0.0:5000
481

482
# With Daphne (Django Channels)
483
# daphne -b 0.0.0.0 -p 5000 main:app
484
```