# aiohttp-asgi-connector
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 AIOHTTP 3.1+ on corresponding compatible Python versions.
## Installation
```sh
$ pdm add aiohttp-asgi-connector
# or
$ python -m pip install --user aiohttp-asgi-connector
```
## Usage
This 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`:
```py
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.
This connector transmits the request to the ASGI application _exactly_ as it is serialized by AIOHTTP. If upload chunking or compression are enabled for your `ClientSession` requests, your ASGI application will need to be able to handle de-chunking and de-compressing; FastAPI / Starlette do not do this by default. Support to enable connector-side dechunking and decompressing may come as a future feature if a need is demonstrated for it (file an Issue).
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](https://pypi.org/project/asgi-lifespan/):
```py
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:
...
```
## License
This software is licensed under the [BSD 3-Clause License](LICENSE).
Raw data
{
"_id": null,
"home_page": null,
"name": "aiohttp-asgi-connector",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "fastapi, aiohttp, asgi, testing, starlette, httpx, asyncio",
"author": null,
"author_email": "thearchitector <me@eliasfgabriel.com>",
"download_url": "https://files.pythonhosted.org/packages/d2/f3/2d28bdd7aa7b1241ef51ae2c9bf4b0480de56f89f501ea1b89b6f78ba265/aiohttp_asgi_connector-1.1.1.tar.gz",
"platform": null,
"description": "# aiohttp-asgi-connector\n\n\n\n\n\nAn AIOHTTP `ClientSession` connector for interacting with ASGI applications.\n\nThis 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.\n\nSupports AIOHTTP 3.1+ on corresponding compatible Python versions.\n\n## Installation\n\n```sh\n$ pdm add aiohttp-asgi-connector\n# or\n$ python -m pip install --user aiohttp-asgi-connector\n```\n\n## Usage\n\nThis library replaces the entire connection stack and underlying HTTP transport. AIOHTTP exposes custom connectors via the `connector` argument supplied when creating a `ClientSession` instance.\n\nTo use the `ASGIApplicationConnector`:\n\n```py\nimport asyncio\nfrom typing import Annotated # or from typing_extensions\n\nfrom aiohttp_asgi_connector import ASGIApplicationConnector\nfrom aiohttp import ClientSession\nfrom fastapi import FastAPI, Body\n\napp = FastAPI()\n\n@app.post(\"/ping\")\nasync def pong(message: Annotated[str, Body(embed=True)]):\n return {\"broadcast\": f\"Application says '{message}'!\"}\n\nasync def main():\n connector = ASGIApplicationConnector(app)\n async with ClientSession(base_url=\"http://localhost\", connector=connector) as session:\n async with session.post(\"/ping\", json={\"message\": \"hello\"}) as resp:\n print(await resp.json())\n # ==> {'broadcast': \"Application says 'hello'!\"}\n\nasyncio.run(main())\n```\n\nExceptions raised within the ASGI application that are not handled by middleware are propagated.\n\nThis connector transmits the request to the ASGI application _exactly_ as it is serialized by AIOHTTP. If upload chunking or compression are enabled for your `ClientSession` requests, your ASGI application will need to be able to handle de-chunking and de-compressing; FastAPI / Starlette do not do this by default. Support to enable connector-side dechunking and decompressing may come as a future feature if a need is demonstrated for it (file an Issue).\n\nThis library does not handle ASGI lifespan events. If you want to run those events, use this library in conjunction with something like [asgi-lifespan](https://pypi.org/project/asgi-lifespan/):\n\n```py\nfrom asgi_lifespan import LifespanManager\n\nasync with LifespanManager(app) as manager:\n connector = ASGIApplicationConnector(manager.app)\n async with ClientSession(base_url=\"http://localhost\", connector=connector) as session:\n ...\n```\n\n## License\n\nThis software is licensed under the [BSD 3-Clause License](LICENSE).\n",
"bugtrack_url": null,
"license": "BSD-3-Clause",
"summary": "AIOHTTP Connector for running ASGI applications",
"version": "1.1.1",
"project_urls": null,
"split_keywords": [
"fastapi",
" aiohttp",
" asgi",
" testing",
" starlette",
" httpx",
" asyncio"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "528b2976aac269aaf0383f534989c79de832a6d408075897656259b7739e33bc",
"md5": "bacb1ac9f8325476e85152ae8c49a8a6",
"sha256": "7692533fe031047eca7123f742fcdea15452c9726303d445a99d9086ef478640"
},
"downloads": -1,
"filename": "aiohttp_asgi_connector-1.1.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "bacb1ac9f8325476e85152ae8c49a8a6",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 7364,
"upload_time": "2025-01-11T08:16:30",
"upload_time_iso_8601": "2025-01-11T08:16:30.771161Z",
"url": "https://files.pythonhosted.org/packages/52/8b/2976aac269aaf0383f534989c79de832a6d408075897656259b7739e33bc/aiohttp_asgi_connector-1.1.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "d2f32d28bdd7aa7b1241ef51ae2c9bf4b0480de56f89f501ea1b89b6f78ba265",
"md5": "d8853a83d0d6a294bfe54c6b8a8f45fc",
"sha256": "4325ee35819b76e4909cc252db4f543e0051e52bed204535cdb2a31e30dcdacd"
},
"downloads": -1,
"filename": "aiohttp_asgi_connector-1.1.1.tar.gz",
"has_sig": false,
"md5_digest": "d8853a83d0d6a294bfe54c6b8a8f45fc",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 6937,
"upload_time": "2025-01-11T08:16:32",
"upload_time_iso_8601": "2025-01-11T08:16:32.879477Z",
"url": "https://files.pythonhosted.org/packages/d2/f3/2d28bdd7aa7b1241ef51ae2c9bf4b0480de56f89f501ea1b89b6f78ba265/aiohttp_asgi_connector-1.1.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-01-11 08:16:32",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "aiohttp-asgi-connector"
}
JSON