[](https://pypi.org/project/aiohttp-deps/)
[](https://pypi.org/project/aiohttp-deps/)
[](https://pypistats.org/packages/aiohttp-deps)
# AioHTTP deps
This project was initially created to show the abillities of [taskiq-dependencies](https://github.com/taskiq-python/taskiq-dependencies) project, which is used by [taskiq](https://github.com/taskiq-python/taskiq) to provide you with the best experience of sending distributed tasks.
This project adds [FastAPI](https://github.com/tiangolo/fastapi)-like dependency injection to your [AioHTTP](https://github.com/aio-libs/aiohttp) application and swagger documentation based on types.
To start using dependency injection, just initialize the injector.
```python
from aiohttp import web
from aiohttp_deps import init as deps_init
app = web.Application()
app.on_startup.append(deps_init)
web.run_app(app)
```
If you use mypy, then we have a custom router with propper types.
```python
from aiohttp import web
from aiohttp_deps import init as deps_init
from aiohttp_deps import Router
router = Router()
@router.get("/")
async def handler():
return web.json_response({})
app = web.Application()
app.router.add_routes(router)
app.on_startup.append(deps_init)
web.run_app(app)
```
Also, you can nest routers with prefixes,
```python
api_router = Router()
memes_router = Router()
main_router = Router()
main_router.add_routes(api_router, prefix="/api")
main_router.add_routes(memes_router, prefix="/memes")
```
## Swagger
If you use dependencies in you handlers, we can easily generate swagger for you.
We have some limitations:
1. We don't support resolving type aliases if hint is a string.
If you define variable like this: `myvar = int | None` and then in handler
you'd create annotation like this: `param: "str | myvar"` it will fail.
You need to unquote type hint in order to get it work.
We will try to fix these limitations later.
To enable swagger, just add it to your startup.
```python
from aiohttp_deps import init, setup_swagger
app = web.Application()
app.on_startup.extend([init, setup_swagger()])
```
### Responses
You can define schema for responses using dataclasses or
pydantic models. This would not affect handlers in any way,
it's only for documentation purposes, if you want to actually
validate values your handler returns, please write your own wrapper.
```python
from dataclasses import dataclass
from aiohttp import web
from pydantic import BaseModel
from aiohttp_deps import Router, openapi_response
router = Router()
@dataclass
class Success:
data: str
class Unauthorized(BaseModel):
why: str
@router.get("/")
@openapi_response(200, Success, content_type="application/xml")
@openapi_response(200, Success)
@openapi_response(401, Unauthorized, description="When token is not correct")
async def handler() -> web.Response:
...
```
This example illustrates how much you can do with this decorator. You
can have multiple content-types for a single status, or you can have different
possble statuses. This function is pretty simple and if you want to make
your own decorator for your responses, it won't be hard.
## Default dependencies
By default this library provides only two injectables. `web.Request` and `web.Application`.
```python
async def handler(app: web.Application = Depends()): ...
async def handler2(request: web.Request = Depends()): ...
```
It's super useful, because you can use these dependencies in
any other dependency. Here's a more complex example of how you can use this library.
```python
from aiohttp_deps import Router, Depends
from aiohttp import web
router = Router()
async def get_db_session(app: web.Application = Depends()):
async with app[web.AppKey("db")] as sess:
yield sess
class MyDAO:
def __init__(self, session=Depends(get_db_session)):
self.session = session
async def get_objects(self) -> list[object]:
return await self.session.execute("SELECT 1")
@router.get("/")
async def handler(db_session: MyDAO = Depends()):
objs = await db_session.get_objects()
return web.json_response({"objects": objs})
```
If you do something like this, you would never think about initializing your DAO. You can just inject it and that's it.
# Built-in dependencies
This library also provides you with some default dependencies that can help you in building the best web-service.
## Json
To parse json, create a pydantic model and add a dependency to your handler.
```python
from aiohttp import web
from pydantic import BaseModel
from aiohttp_deps import Router, Json, Depends
router = Router()
class UserInfo(BaseModel):
name: str
@router.post("/users")
async def new_data(user: UserInfo = Depends(Json())):
return web.json_response({"user": user.model_dump()})
```
This dependency automatically validates data and send
errors if the data doesn't orrelate with schema or body is not a valid json.
If you want to make this data optional, just mark it as optional.
```python
@router.post("/users")
async def new_data(user: Optional[UserInfo] = Depends(Json())):
if user is None:
return web.json_response({"user": None})
return web.json_response({"user": user.model_dump()})
```
## Headers
You can get and validate headers using `Header` dependency.
Let's try to build simple example for authorization.
```python
from aiohttp_deps import Router, Header, Depends
from aiohttp import web
router = Router()
def decode_token(authorization: str = Depends(Header())) -> str:
if authorization == "secret":
# Let's pretend that here we
# decode our token.
return authorization
raise web.HTTPUnauthorized()
@router.get("/secret_data")
async def new_data(token: str = Depends(decode_token)) -> web.Response:
return web.json_response({"secret": "not a secret"})
```
As you can see, header name to parse is equal to the
name of a parameter that introduces Header dependency.
If you want to use some name that is not allowed in python, or just want to have different names, you can use alias. Like this:
```python
def decode_token(auth: str = Depends(Header(alias="Authorization"))) -> str:
```
Headers can also be parsed to types. If you want a header to be parsed as int, just add the typehint.
```python
def decode_token(meme_id: int = Depends(Header())) -> str:
```
If you want to get list of values of one header, use parameter `multiple=True`.
```python
def decode_token(meme_id: list[int] = Depends(Header(multiple=True))) -> str:
```
And, of course, you can provide this dependency with default value if the value from user cannot be parsed for some reason.
```python
def decode_token(meme_id: str = Depends(Header(default="not-a-secret"))) -> str:
```
## Queries
You can depend on `Query` to get and parse query parameters.
```python
from aiohttp_deps import Router, Query, Depends
from aiohttp import web
router = Router()
@router.get("/shop")
async def shop(item_id: str = Depends(Query())) -> web.Response:
return web.json_response({"id": item_id})
```
the name of the parameter is the same as the name of function parameter.
The Query dependency is acually the same as the Header dependency, so everything about the `Header` dependency also applies to `Query`.
## Views
If you use views as handlers, please use View class from `aiohttp_deps`, otherwise the magic won't work.
```python
from aiohttp_deps import Router, View, Depends
from aiohttp import web
router = Router()
@router.view("/view")
class MyView(View):
async def get(self, app: web.Application = Depends()):
return web.json_response({"app": str(app)})
```
## Forms
Now you can easiy get and validate form data from your request.
To make the magic happen, please add `arbitrary_types_allowed` to the config of your model.
```python
import pydantic
from aiohttp_deps import Router, Depends, Form
from aiohttp import web
router = Router()
class MyForm(pydantic.BaseModel):
id: int
file: web.FileField
model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)
@router.post("/")
async def handler(my_form: MyForm = Depends(Form())):
with open("my_file", "wb") as f:
f.write(my_form.file.file.read())
return web.json_response({"id": my_form.id})
```
## Path
If you have path variables, you can also inject them in your handler.
```python
from aiohttp_deps import Router, Path, Depends
from aiohttp import web
router = Router()
@router.get("/view/{var}")
async def my_handler(var: str = Depends(Path())):
return web.json_response({"var": var})
```
## Overridiing dependencies
Sometimes for tests you don't want to calculate actual functions
and you want to pass another functions instead.
To do so, you can add "dependency_overrides" or "values_overrides" to the aplication's state.
These values should be dicts. The keys for these values can be found in `aiohttp_deps.keys` module.
Here's an example.
```python
def original_dep() -> int:
return 1
class MyView(View):
async def get(self, num: int = Depends(original_dep)):
"""Nothing."""
return web.json_response({"request": num})
```
Imagine you have a handler that depends on some function,
but instead of `1` you want to have `2` in your tests.
To do it, jsut add `dependency_overrides` somewhere,
where you create your application. And make sure that keys
of that dict are actual function that are being replaced.
```python
from aiohttp_deps import VALUES_OVERRIDES_KEY
my_app[VALUES_OVERRIDES_KEY] = {original_dep: 2}
```
But `values_overrides` only overrides returned values. If you want to
override functions, you have to use `dependency_overrides`. Here's an example:
```python
from aiohttp_deps import DEPENDENCY_OVERRIDES_KEY
def replacing_function() -> int:
return 2
my_app[DEPENDENCY_OVERRIDES_KEY] = {original_dep: replacing_function}
```
The cool point about `dependency_overrides`, is that it recalculates graph and
you can use dependencies in function that replaces the original.
Raw data
{
"_id": null,
"home_page": "https://github.com/taskiq-python/aiohttp-deps",
"name": "aiohttp-deps",
"maintainer": "Taskiq team",
"docs_url": null,
"requires_python": ">=3.8.1,<4.0.0",
"maintainer_email": "taskiq@no-reply.com",
"keywords": "aiohttp,taskiq-dependencies",
"author": "Taskiq team",
"author_email": "taskiq@no-reply.com",
"download_url": "https://files.pythonhosted.org/packages/5f/57/1614601a7cf65372f441d6f9dbd17c361cabaee62ca1efc438297ade2d22/aiohttp_deps-1.1.2.tar.gz",
"platform": null,
"description": "[](https://pypi.org/project/aiohttp-deps/)\n[](https://pypi.org/project/aiohttp-deps/)\n[](https://pypistats.org/packages/aiohttp-deps)\n\n# AioHTTP deps\n\n\nThis project was initially created to show the abillities of [taskiq-dependencies](https://github.com/taskiq-python/taskiq-dependencies) project, which is used by [taskiq](https://github.com/taskiq-python/taskiq) to provide you with the best experience of sending distributed tasks.\n\nThis project adds [FastAPI](https://github.com/tiangolo/fastapi)-like dependency injection to your [AioHTTP](https://github.com/aio-libs/aiohttp) application and swagger documentation based on types.\n\nTo start using dependency injection, just initialize the injector.\n\n```python\nfrom aiohttp import web\nfrom aiohttp_deps import init as deps_init\n\n\napp = web.Application()\n\n\napp.on_startup.append(deps_init)\n\nweb.run_app(app)\n\n```\n\n\nIf you use mypy, then we have a custom router with propper types.\n\n\n```python\nfrom aiohttp import web\nfrom aiohttp_deps import init as deps_init\nfrom aiohttp_deps import Router\n\nrouter = Router()\n\n\n@router.get(\"/\")\nasync def handler():\n return web.json_response({})\n\n\napp = web.Application()\n\napp.router.add_routes(router)\n\napp.on_startup.append(deps_init)\n\nweb.run_app(app)\n\n```\n\nAlso, you can nest routers with prefixes,\n\n```python\napi_router = Router()\n\nmemes_router = Router()\n\nmain_router = Router()\n\nmain_router.add_routes(api_router, prefix=\"/api\")\nmain_router.add_routes(memes_router, prefix=\"/memes\")\n```\n\n## Swagger\n\nIf you use dependencies in you handlers, we can easily generate swagger for you.\nWe have some limitations:\n1. We don't support resolving type aliases if hint is a string.\n If you define variable like this: `myvar = int | None` and then in handler\n you'd create annotation like this: `param: \"str | myvar\"` it will fail.\n You need to unquote type hint in order to get it work.\n\nWe will try to fix these limitations later.\n\nTo enable swagger, just add it to your startup.\n\n```python\nfrom aiohttp_deps import init, setup_swagger\n\napp = web.Application()\n\napp.on_startup.extend([init, setup_swagger()])\n```\n\n### Responses\n\nYou can define schema for responses using dataclasses or\npydantic models. This would not affect handlers in any way,\nit's only for documentation purposes, if you want to actually\nvalidate values your handler returns, please write your own wrapper.\n\n```python\nfrom dataclasses import dataclass\n\nfrom aiohttp import web\nfrom pydantic import BaseModel\n\nfrom aiohttp_deps import Router, openapi_response\n\nrouter = Router()\n\n\n@dataclass\nclass Success:\n data: str\n\n\nclass Unauthorized(BaseModel):\n why: str\n\n\n@router.get(\"/\")\n@openapi_response(200, Success, content_type=\"application/xml\")\n@openapi_response(200, Success)\n@openapi_response(401, Unauthorized, description=\"When token is not correct\")\nasync def handler() -> web.Response:\n ...\n```\n\nThis example illustrates how much you can do with this decorator. You\ncan have multiple content-types for a single status, or you can have different\npossble statuses. This function is pretty simple and if you want to make\nyour own decorator for your responses, it won't be hard.\n\n\n## Default dependencies\n\nBy default this library provides only two injectables. `web.Request` and `web.Application`.\n\n```python\n\nasync def handler(app: web.Application = Depends()): ...\n\nasync def handler2(request: web.Request = Depends()): ...\n\n```\n\nIt's super useful, because you can use these dependencies in\nany other dependency. Here's a more complex example of how you can use this library.\n\n\n```python\nfrom aiohttp_deps import Router, Depends\nfrom aiohttp import web\n\nrouter = Router()\n\n\nasync def get_db_session(app: web.Application = Depends()):\n async with app[web.AppKey(\"db\")] as sess:\n yield sess\n\n\nclass MyDAO:\n def __init__(self, session=Depends(get_db_session)):\n self.session = session\n\n async def get_objects(self) -> list[object]:\n return await self.session.execute(\"SELECT 1\")\n\n\n@router.get(\"/\")\nasync def handler(db_session: MyDAO = Depends()):\n objs = await db_session.get_objects()\n return web.json_response({\"objects\": objs})\n\n```\n\nIf you do something like this, you would never think about initializing your DAO. You can just inject it and that's it.\n\n\n# Built-in dependencies\n\nThis library also provides you with some default dependencies that can help you in building the best web-service.\n\n## Json\n\nTo parse json, create a pydantic model and add a dependency to your handler.\n\n\n```python\nfrom aiohttp import web\nfrom pydantic import BaseModel\nfrom aiohttp_deps import Router, Json, Depends\n\nrouter = Router()\n\n\nclass UserInfo(BaseModel):\n name: str\n\n\n@router.post(\"/users\")\nasync def new_data(user: UserInfo = Depends(Json())):\n return web.json_response({\"user\": user.model_dump()})\n\n```\n\nThis dependency automatically validates data and send\nerrors if the data doesn't orrelate with schema or body is not a valid json.\n\nIf you want to make this data optional, just mark it as optional.\n\n```python\n@router.post(\"/users\")\nasync def new_data(user: Optional[UserInfo] = Depends(Json())):\n if user is None:\n return web.json_response({\"user\": None})\n return web.json_response({\"user\": user.model_dump()})\n\n```\n\n## Headers\n\nYou can get and validate headers using `Header` dependency.\n\nLet's try to build simple example for authorization.\n\n```python\nfrom aiohttp_deps import Router, Header, Depends\nfrom aiohttp import web\n\nrouter = Router()\n\n\ndef decode_token(authorization: str = Depends(Header())) -> str:\n if authorization == \"secret\":\n # Let's pretend that here we\n # decode our token.\n return authorization\n raise web.HTTPUnauthorized()\n\n\n@router.get(\"/secret_data\")\nasync def new_data(token: str = Depends(decode_token)) -> web.Response:\n return web.json_response({\"secret\": \"not a secret\"})\n\n```\n\nAs you can see, header name to parse is equal to the\nname of a parameter that introduces Header dependency.\n\nIf you want to use some name that is not allowed in python, or just want to have different names, you can use alias. Like this:\n\n```python\ndef decode_token(auth: str = Depends(Header(alias=\"Authorization\"))) -> str:\n```\n\nHeaders can also be parsed to types. If you want a header to be parsed as int, just add the typehint.\n\n```python\ndef decode_token(meme_id: int = Depends(Header())) -> str:\n```\n\nIf you want to get list of values of one header, use parameter `multiple=True`.\n\n```python\ndef decode_token(meme_id: list[int] = Depends(Header(multiple=True))) -> str:\n\n```\n\nAnd, of course, you can provide this dependency with default value if the value from user cannot be parsed for some reason.\n\n```python\ndef decode_token(meme_id: str = Depends(Header(default=\"not-a-secret\"))) -> str:\n```\n\n\n## Queries\n\nYou can depend on `Query` to get and parse query parameters.\n\n```python\nfrom aiohttp_deps import Router, Query, Depends\nfrom aiohttp import web\n\nrouter = Router()\n\n\n@router.get(\"/shop\")\nasync def shop(item_id: str = Depends(Query())) -> web.Response:\n return web.json_response({\"id\": item_id})\n\n```\n\nthe name of the parameter is the same as the name of function parameter.\n\nThe Query dependency is acually the same as the Header dependency, so everything about the `Header` dependency also applies to `Query`.\n\n## Views\n\nIf you use views as handlers, please use View class from `aiohttp_deps`, otherwise the magic won't work.\n\n```python\nfrom aiohttp_deps import Router, View, Depends\nfrom aiohttp import web\n\nrouter = Router()\n\n\n@router.view(\"/view\")\nclass MyView(View):\n async def get(self, app: web.Application = Depends()):\n return web.json_response({\"app\": str(app)})\n\n```\n\n\n## Forms\n\nNow you can easiy get and validate form data from your request.\nTo make the magic happen, please add `arbitrary_types_allowed` to the config of your model.\n\n\n```python\nimport pydantic\nfrom aiohttp_deps import Router, Depends, Form\nfrom aiohttp import web\n\nrouter = Router()\n\n\nclass MyForm(pydantic.BaseModel):\n id: int\n file: web.FileField\n\n model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)\n\n\n@router.post(\"/\")\nasync def handler(my_form: MyForm = Depends(Form())):\n with open(\"my_file\", \"wb\") as f:\n f.write(my_form.file.file.read())\n return web.json_response({\"id\": my_form.id})\n\n```\n\n## Path\n\nIf you have path variables, you can also inject them in your handler.\n\n```python\nfrom aiohttp_deps import Router, Path, Depends\nfrom aiohttp import web\n\nrouter = Router()\n\n\n@router.get(\"/view/{var}\")\nasync def my_handler(var: str = Depends(Path())):\n return web.json_response({\"var\": var})\n\n```\n\n\n## Overridiing dependencies\n\nSometimes for tests you don't want to calculate actual functions\nand you want to pass another functions instead.\n\nTo do so, you can add \"dependency_overrides\" or \"values_overrides\" to the aplication's state.\nThese values should be dicts. The keys for these values can be found in `aiohttp_deps.keys` module.\n\nHere's an example.\n\n```python\ndef original_dep() -> int:\n return 1\n\nclass MyView(View):\n async def get(self, num: int = Depends(original_dep)):\n \"\"\"Nothing.\"\"\"\n return web.json_response({\"request\": num})\n```\n\nImagine you have a handler that depends on some function,\nbut instead of `1` you want to have `2` in your tests.\n\nTo do it, jsut add `dependency_overrides` somewhere,\nwhere you create your application. And make sure that keys\nof that dict are actual function that are being replaced.\n\n```python\nfrom aiohttp_deps import VALUES_OVERRIDES_KEY\n\nmy_app[VALUES_OVERRIDES_KEY] = {original_dep: 2}\n```\n\nBut `values_overrides` only overrides returned values. If you want to\noverride functions, you have to use `dependency_overrides`. Here's an example:\n\n```python\nfrom aiohttp_deps import DEPENDENCY_OVERRIDES_KEY\n\ndef replacing_function() -> int:\n return 2\n\n\nmy_app[DEPENDENCY_OVERRIDES_KEY] = {original_dep: replacing_function}\n```\n\nThe cool point about `dependency_overrides`, is that it recalculates graph and\nyou can use dependencies in function that replaces the original.\n",
"bugtrack_url": null,
"license": "LICENSE",
"summary": "Dependency injection for AioHTTP",
"version": "1.1.2",
"project_urls": {
"Homepage": "https://github.com/taskiq-python/aiohttp-deps"
},
"split_keywords": [
"aiohttp",
"taskiq-dependencies"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "bc8fdeef06ea5307e75ea59797aa46357c33aaec08bf40c7f86c924f721b174c",
"md5": "d0a26fac276cca5bf9ab41ae26f22b9a",
"sha256": "5e4dfe1d06a00dd5907a557a0b3c9b2df70f91fe00d29b6213439a96f37ca3dd"
},
"downloads": -1,
"filename": "aiohttp_deps-1.1.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "d0a26fac276cca5bf9ab41ae26f22b9a",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8.1,<4.0.0",
"size": 14881,
"upload_time": "2024-02-05T20:25:57",
"upload_time_iso_8601": "2024-02-05T20:25:57.520799Z",
"url": "https://files.pythonhosted.org/packages/bc/8f/deef06ea5307e75ea59797aa46357c33aaec08bf40c7f86c924f721b174c/aiohttp_deps-1.1.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "5f571614601a7cf65372f441d6f9dbd17c361cabaee62ca1efc438297ade2d22",
"md5": "65e27213d2cdca3728eefec49daa48e0",
"sha256": "b8e44123c4370f03b07fab471e693269bb2d2fc946ad89403b5af91506e1651a"
},
"downloads": -1,
"filename": "aiohttp_deps-1.1.2.tar.gz",
"has_sig": false,
"md5_digest": "65e27213d2cdca3728eefec49daa48e0",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8.1,<4.0.0",
"size": 16014,
"upload_time": "2024-02-05T20:25:59",
"upload_time_iso_8601": "2024-02-05T20:25:59.319930Z",
"url": "https://files.pythonhosted.org/packages/5f/57/1614601a7cf65372f441d6f9dbd17c361cabaee62ca1efc438297ade2d22/aiohttp_deps-1.1.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-02-05 20:25:59",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "taskiq-python",
"github_project": "aiohttp-deps",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"tox": true,
"lcname": "aiohttp-deps"
}
JSON
