# 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 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 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.
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](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/e1/48/8eddcbfd7c9c5bab3497d2a48a9b2703e4aca83d938942d190d1359dd6e3/aiohttp_asgi_connector-1.0.3.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 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 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\nSince no HTTP packets are actually sent, request chunking and compression have no effect when used.\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.0.3",
"project_urls": null,
"split_keywords": [
"fastapi",
" aiohttp",
" asgi",
" testing",
" starlette",
" httpx",
" asyncio"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "bf0b2719872adcca627f994c10bc6e0a85ec4398b780b29742fd0d44305b1091",
"md5": "0d838a8976323b3d03a8fb1c88cccdc3",
"sha256": "1ddeda8e67cdb6e24f7b2b27ed14efc4fdccecf7e77924050a33d6a0ba6eedd2"
},
"downloads": -1,
"filename": "aiohttp_asgi_connector-1.0.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "0d838a8976323b3d03a8fb1c88cccdc3",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 6776,
"upload_time": "2024-08-28T04:05:30",
"upload_time_iso_8601": "2024-08-28T04:05:30.427360Z",
"url": "https://files.pythonhosted.org/packages/bf/0b/2719872adcca627f994c10bc6e0a85ec4398b780b29742fd0d44305b1091/aiohttp_asgi_connector-1.0.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "e1488eddcbfd7c9c5bab3497d2a48a9b2703e4aca83d938942d190d1359dd6e3",
"md5": "91290ff3d9abd42244ff175fc6d68e44",
"sha256": "313e6c4d4d887757d213aac28379e2fa0020bb187a72e5382bd79f2c0ee4802e"
},
"downloads": -1,
"filename": "aiohttp_asgi_connector-1.0.3.tar.gz",
"has_sig": false,
"md5_digest": "91290ff3d9abd42244ff175fc6d68e44",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 6554,
"upload_time": "2024-08-28T04:05:31",
"upload_time_iso_8601": "2024-08-28T04:05:31.714522Z",
"url": "https://files.pythonhosted.org/packages/e1/48/8eddcbfd7c9c5bab3497d2a48a9b2703e4aca83d938942d190d1359dd6e3/aiohttp_asgi_connector-1.0.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-08-28 04:05:31",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "aiohttp-asgi-connector"
}
JSON