asyncutilsx

ASGI wrapper for combining FastAPI and Socket.IO in one app.

Author: Akshat kotpalliwar (alias IntegerAlex)
SPDX-License-Identifier: LGPL-2.1-only

Minimal and composable: one core function, isolated effects at the ASGI boundary.

Breaking changes (v0.3.0)

health_check_route has been removed from the public API. Add your own /health endpoint directly in your FastAPI app instead. See docs/user_guide.md for details.

Why asyncplus() / asyncutilsx?

Scenario: real-time chat app with REST API

# ❌ With mount() — problematic
app = FastAPI()
app.mount("/socket.io", socket_app)
# - Auth middleware may break Socket.IO handshake
# - CORS middleware may interfere
# - Extra latency on every Socket.IO message

# ✅ With asyncplus — clean
asgi_app = asyncplus(app, sio)
# - Socket.IO gets raw ASGI requests
# - FastAPI gets HTTP requests
# - Each handles its own concerns

FastAPI’s app.mount("/path", other_asgi) works, but you must serve Socket.IO on a subpath and deal with that path everywhere (client, CORS, proxies). asyncplus gives you a single ASGI app: one entry point for the server (e.g. uvicorn), no mount path, same origin for API and Socket.IO. Plug and play—no extra middleware added; you keep full control of the ASGI apps you pass in.

Install

pip install asyncutilsx

Usage

from fastapi import FastAPI
from socketio import AsyncServer
from asyncutilsx import asyncplus, create_app

app = FastAPI()
sio = AsyncServer(async_mode="asgi")

@sio.event
async def connect(sid, environ):
    print("connect", sid)

asgi_app = asyncplus(app, sio)
# Run with: uvicorn asgi:asgi_app

Or use the convenience helper: asgi_app = create_app(app, sio).

• HTTP (except /socket.io/*) → FastAPI
• HTTP /socket.io/* → Socket.IO (polling)
• WebSocket (path matching socketio_path) → Socket.IO; other WebSocket paths → FastAPI
• Lifespan → both apps (startup/shutdown events are multiplexed so each app boots and shuts down properly)

Optional: asyncplus(app, sio, socketio_path="/custom/", debug_hook=..., socketio_fallback_on_error=False, timeout=30.0).

For custom routing (e.g. SSE, gRPC), use the pure router(routes, default_app) function: pass a sequence of (predicate, app) pairs; first match wins; default_app used when none match.

See docs/user_guide.md for common patterns and examples.

Development

pip install -e ".[dev]"
pytest

Benchmarks

Performance benchmarks are continuously monitored with CodSpeed. To run benchmarks locally:

pytest benchmarks/ --codspeed

asyncplus vs FastAPI mount

The suite includes direct comparisons between asyncplus and the FastAPI mount workaround (app.mount("/socket.io", socket_app)) on the same scenarios:

Same scopes and mocks are used for both approaches so results are comparable. See benchmarks/test_bench_asyncplus_vs_mount.py.

License

LGPL-2.1-only