An AIOHTTP ClientSession connector for interacting with ASGI applications.
This library intends to increase the parity between AIOHTTP and HTTPX, specifically with HTTPX's AsyncClient. It is primarily intended to be used in test suite scenarios, or other situations where one would want to interface with an ASGI application directly instead of through a web server.
Supports Python 3.8+ and AIOHTTP 3+.
$ pdm add aiohttp-asgi-connector
# or
$ python -m pip install --user aiohttp-asgi-connectorThis library replaces the entire connection stack and underlying HTTP transport. AIOHTTP exposes custom connectors via the connector argument supplied when creating a ClientSession instance.
To use the ASGIApplicationConnector:
import asyncio
from typing import Annotated # or from typing_extensions
from aiohttp_asgi_connector import ASGIApplicationConnector
from aiohttp import ClientSession
from fastapi import FastAPI, Body
app = FastAPI()
@app.post("/ping")
async def pong(message: Annotated[str, Body(embed=True)]):
return {"broadcast": f"Application says '{message}'!"}
async def main():
connector = ASGIApplicationConnector(app)
async with ClientSession(base_url="http://localhost", connector=connector) as session:
async with session.post("/ping", json={"message": "hello"}) as resp:
print(await resp.json())
# ==> {'broadcast': "Application says 'hello'!"}
asyncio.run(main())Exceptions raised within the ASGI application that are not handled by middleware are propagated.
Since no HTTP packets are actually sent, request chunking and compression have no effect when used.
This library does not handle ASGI lifespan events. If you want to run those events, use this library in conjunction with something like asgi-lifespan:
from asgi_lifespan import LifespanManager
async with LifespanManager(app) as manager:
connector = ASGIApplicationConnector(manager.app)
async with ClientSession(base_url="http://localhost", connector=connector) as session:
...This software is licensed under the BSD 3-Clause License.