# aiohttp-asgi-connector
![GitHub Workflow Status](https://img.shields.io/github/actions/workflow/status/thearchitector/aiohttp-asgi-connector/CI.yaml?label=tests&style=flat-square)
![PyPI - Downloads](https://img.shields.io/pypi/dw/aiohttp-asgi-connector?style=flat-square)
![GitHub](https://img.shields.io/github/license/thearchitector/aiohttp-asgi-connector?style=flat-square)
AIOHTTP Connector for running 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+.
## 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 pool 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 the application, like `HTTPException` in FastAPI for example, are propagated.
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/).
## 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/1c/92/38c93781b95055a4b80618ea2a7b18f83bf0c1033bc26557cc355ecf181d/aiohttp_asgi_connector-1.0.2.tar.gz",
"platform": null,
"description": "# aiohttp-asgi-connector\n\n![GitHub Workflow Status](https://img.shields.io/github/actions/workflow/status/thearchitector/aiohttp-asgi-connector/CI.yaml?label=tests&style=flat-square)\n![PyPI - Downloads](https://img.shields.io/pypi/dw/aiohttp-asgi-connector?style=flat-square)\n![GitHub](https://img.shields.io/github/license/thearchitector/aiohttp-asgi-connector?style=flat-square)\n\nAIOHTTP Connector for running 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 Python 3.8+ and AIOHTTP 3+.\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 pool 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 the application, like `HTTPException` in FastAPI for example, are propagated.\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## 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.0.2",
"project_urls": null,
"split_keywords": [
"fastapi",
" aiohttp",
" asgi",
" testing",
" starlette",
" httpx",
" asyncio"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "e5fa30171e3229b0f4b4ae557414caea5ca7c07183bfb8efb83759b36b093e52",
"md5": "0a071c1037823560b415313cac40d88c",
"sha256": "5c5878eb829391be2b4a4ab88e21c57f59c42f61ed7a601d49856250e80697d9"
},
"downloads": -1,
"filename": "aiohttp_asgi_connector-1.0.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "0a071c1037823560b415313cac40d88c",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 6592,
"upload_time": "2024-06-29T07:14:05",
"upload_time_iso_8601": "2024-06-29T07:14:05.023252Z",
"url": "https://files.pythonhosted.org/packages/e5/fa/30171e3229b0f4b4ae557414caea5ca7c07183bfb8efb83759b36b093e52/aiohttp_asgi_connector-1.0.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "1c9238c93781b95055a4b80618ea2a7b18f83bf0c1033bc26557cc355ecf181d",
"md5": "453013a59aa77e549ab2e6ec2c37e2e5",
"sha256": "3d1bea0382a8b0867a3f307b236d6049d0b54c56c406b85211873f30afb0c306"
},
"downloads": -1,
"filename": "aiohttp_asgi_connector-1.0.2.tar.gz",
"has_sig": false,
"md5_digest": "453013a59aa77e549ab2e6ec2c37e2e5",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 6157,
"upload_time": "2024-06-29T07:14:06",
"upload_time_iso_8601": "2024-06-29T07:14:06.512373Z",
"url": "https://files.pythonhosted.org/packages/1c/92/38c93781b95055a4b80618ea2a7b18f83bf0c1033bc26557cc355ecf181d/aiohttp_asgi_connector-1.0.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-06-29 07:14:06",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "aiohttp-asgi-connector"
}