Dependency injection for AioHTTP
Project description
AioHTTP deps
This project was initially created to show the abillities of taskiq-dependencies project, which is used by taskiq to provide you with the best experience of sending distributed tasks.
This project adds FastAPI-like dependency injection to your AioHTTP application.
To start using dependency injection, just initialize the injector.
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.
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)
Default dependencies
By default this library provides only two injectables. web.Request
and web.Application
.
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.
from aiohttp_deps import Router, Depends
from aiohttp import web
router = Router()
async def get_db_session(app: web.Application = Depends()):
async with app["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.
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.dict()})
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.
@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.dict()})
Headers
You can get and validate headers using Header
dependency.
Let's try to build simple example for authorization.
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:
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.
def decode_token(meme_id: int = Depends(Header())) -> str:
If you want to get list of values of one header, use parameter multiple=True
.
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.
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.
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.
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)})
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for aiohttp_deps-0.1.1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 48dc074c8da58c852fb70a4ff6fba2759135201133209a21493b2c85cbf20710 |
|
MD5 | fa7204f4b2390392fc8a0252925c74c0 |
|
BLAKE2b-256 | 42978902b0c11076c322b7e95c38255e065ef195020d4e854c9bda2ed2f347cb |