AioHTTP deps

This project was initially created to show the abilities 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 and swagger documentation based on types.

To start using dependency injection, just initialize the injector.

fromaiohttpimportwebfromaiohttp_depsimportinitasdeps_initapp=web.Application()
app.on_startup.append(deps_init)
web.run_app(app)

If you use mypy, then we have a custom router with proper types.

fromaiohttpimportwebfromaiohttp_depsimportinitasdeps_initfromaiohttp_depsimportRouterrouter=Router()
@router.get("/")asyncdefhandler():
returnweb.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,

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.

fromaiohttp_depsimportinit, setup_swaggerapp=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.

fromdataclassesimportdataclassfromaiohttpimportwebfrompydanticimportBaseModelfromaiohttp_depsimportRouter, openapi_responserouter=Router()
@dataclassclassSuccess:
data: strclassUnauthorized(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")asyncdefhandler() ->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 possible 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.

asyncdefhandler(app: web.Application=Depends()): ...
asyncdefhandler2(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.

fromaiohttp_depsimportRouter, Dependsfromaiohttpimportwebrouter=Router()
asyncdefget_db_session(app: web.Application=Depends()):
asyncwithapp[web.AppKey("db")] assess:
yieldsessclassMyDAO:
def__init__(self, session=Depends(get_db_session)):
self.session=sessionasyncdefget_objects(self) ->list[object]:
returnawaitself.session.execute("SELECT 1")
@router.get("/")asyncdefhandler(db_session: MyDAO=Depends()):
objs=awaitdb_session.get_objects()
returnweb.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.

fromaiohttpimportwebfrompydanticimportBaseModelfromaiohttp_depsimportRouter, Json, Dependsrouter=Router()
classUserInfo(BaseModel):
name: str@router.post("/users")asyncdefnew_data(user: UserInfo=Depends(Json())):
returnweb.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.

@router.post("/users")asyncdefnew_data(user: Optional[UserInfo] =Depends(Json())):
ifuserisNone:
returnweb.json_response({"user": None})
returnweb.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.

fromaiohttp_depsimportRouter, Header, Dependsfromaiohttpimportwebrouter=Router()
defdecode_token(authorization: str=Depends(Header())) ->str:
ifauthorization=="secret":
# Let's pretend that here we# decode our token.returnauthorizationraiseweb.HTTPUnauthorized()
@router.get("/secret_data")asyncdefnew_data(token: str=Depends(decode_token)) ->web.Response:
returnweb.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:

defdecode_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.

defdecode_token(meme_id: int=Depends(Header())) ->str:

If you want to get list of values of one header, use parameter multiple=True.

defdecode_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.

defdecode_token(meme_id: str=Depends(Header(default="not-a-secret"))) ->str:

Queries

You can depend on Query to get and parse query parameters.

fromaiohttp_depsimportRouter, Query, Dependsfromaiohttpimportwebrouter=Router()
@router.get("/shop")asyncdefshop(item_id: str=Depends(Query())) ->web.Response:
returnweb.json_response({"id": item_id})

the name of the parameter is the same as the name of function parameter.

The Query dependency is actually 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.

fromaiohttp_depsimportRouter, View, Dependsfromaiohttpimportwebrouter=Router()
@router.view("/view")classMyView(View):
asyncdefget(self, app: web.Application=Depends()):
returnweb.json_response({"app": str(app)})

Forms

Now you can easily get and validate form data from your request. To make the magic happen, please add arbitrary_types_allowed to the config of your model.

importpydanticfromaiohttp_depsimportRouter, Depends, Formfromaiohttpimportwebrouter=Router()
classMyForm(pydantic.BaseModel):
id: intfile: web.FileFieldmodel_config=pydantic.ConfigDict(arbitrary_types_allowed=True)
@router.post("/")asyncdefhandler(my_form: MyForm=Depends(Form())):
withopen("my_file", "wb") asf:
f.write(my_form.file.file.read())
returnweb.json_response({"id": my_form.id})

Path

If you have path variables, you can also inject them in your handler.

fromaiohttp_depsimportRouter, Path, Dependsfromaiohttpimportwebrouter=Router()
@router.get("/view/{var}")asyncdefmy_handler(var: str=Depends(Path())):
returnweb.json_response({"var": var})

ExtraOpenAPI

This dependency is used to add additional swagger fields to the endpoint's swagger that is using this dependency. It might be even indirect dependency.

You can check how this thing can be used in our examples/swagger_auth.py.

Overriding 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 application's state. These values should be dicts. The keys for these values can be found in aiohttp_deps.keys module.

Here's an example.

deforiginal_dep() ->int:
return1classMyView(View):
asyncdefget(self, num: int=Depends(original_dep)):
"""Nothing."""returnweb.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, just add dependency_overrides somewhere, where you create your application. And make sure that keys of that dict are actual function that are being replaced.

fromaiohttp_depsimportVALUES_OVERRIDES_KEYmy_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:

fromaiohttp_depsimportDEPENDENCY_OVERRIDES_KEYdefreplacing_function() ->int:
return2my_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.