aiohttp-asgi-connector

raw JSON →
1.1.2 verified Fri May 01 auth: no python

Provides an aiohttp connector for running ASGI applications (e.g., Starlette, FastAPI) inside an aiohttp test harness or client without needing a live server. Current version 1.1.2, supports Python >=3.8, maintained on GitHub.

pip install aiohttp-asgi-connector
error ImportError: cannot import name 'ASGIConnector' from 'aiohttp'
cause Attempting to import ASGIConnector from the aiohttp package, which doesn't expose it.
fix
Install aiohttp-asgi-connector and use: from aiohttp_asgi_connector import ASGIConnector
error AttributeError: 'NoneType' object has no attribute 'send'
cause ASGI app does not have a lifespan handler or is not properly handling lifespan messages. Often happens with simple apps that lack lifespan.
fix
Pass lifespan=False to ASGIConnector: connector = ASGIConnector(app, lifespan=False)
error RuntimeError: Event loop is closed
cause Running an async test outside an event loop or incorrectly closing the loop before the session finishes.
fix
Use asyncio.run() or an async test framework like pytest-asyncio to manage the loop.
gotcha The ASGI app must have a lifespan handler. If the app does not implement lifespan events (startup/shutdown), you may need to pass 'lifespan=False' to ASGIConnector.
fix connector = ASGIConnector(app, lifespan=False)
gotcha The connector automatically adds a 'testserver' host header. If your ASGI app validates the Host header, ensure it matches 'testserver' or disable host checking.
fix Override the Host header in your aiohttp request or configure the ASGI app accordingly.
deprecated Older versions required importing from 'aiohttp_asgi' (note: singular 'asgi'). The package name changed to 'aiohttp-asgi-connector' and import path to 'aiohttp_asgi_connector'.
fix Update install and import to 'aiohttp-asgi-connector' and 'from aiohttp_asgi_connector import ASGIConnector'.

Creates an aiohttp connector that wraps an ASGI app for in-process testing.

import asyncio
import aiohttp
from aiohttp_asgi_connector import ASGIConnector
from starlette.applications import Starlette
from starlette.responses import PlainTextResponse

async def test():
    app = Starlette()
    async def homepage(request):
        return PlainTextResponse('Hello, World!')
    app.add_route('/', homepage)

    connector = ASGIConnector(app)
    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get('http://testserver/') as resp:
            print(await resp.text())

asyncio.run(test())