|
1 | | -# aiohttp-deps |
| 1 | +[](https://pypi.org/project/aiohttp-deps/) |
| 2 | +[](https://pypi.org/project/aiohttp-deps/) |
| 3 | +[](https://pypistats.org/packages/aiohttp-deps) |
| 4 | + |
| 5 | +# AioHTTP deps |
| 6 | + |
| 7 | + |
| 8 | +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. |
| 9 | + |
| 10 | +This project adds [FastAPI](https://github.com/tiangolo/fastapi)-like dependency injection to your [AioHTTP](https://github.com/aio-libs/aiohttp) application. |
| 11 | + |
| 12 | +To start using dependency injection, just initialize the injector. |
| 13 | + |
| 14 | +```python |
| 15 | +from aiohttp import web |
| 16 | +from aiohttp_deps import init as deps_init |
| 17 | + |
| 18 | + |
| 19 | +app = web.Application() |
| 20 | + |
| 21 | + |
| 22 | +app.on_startup.append(deps_init) |
| 23 | + |
| 24 | +web.run_app(app) |
| 25 | + |
| 26 | +``` |
| 27 | + |
| 28 | + |
| 29 | +If you use mypy, then we have a custom router with propper types. |
| 30 | + |
| 31 | + |
| 32 | +```python |
| 33 | +from aiohttp import web |
| 34 | +from aiohttp_deps import init as deps_init |
| 35 | +from aiohttp_deps import Router |
| 36 | + |
| 37 | +router = Router() |
| 38 | + |
| 39 | + |
| 40 | +@router.get("/") |
| 41 | +async def handler(): |
| 42 | + return web.json_response({}) |
| 43 | + |
| 44 | + |
| 45 | +app = web.Application() |
| 46 | + |
| 47 | +app.router.add_routes(router) |
| 48 | + |
| 49 | +app.on_startup.append(deps_init) |
| 50 | + |
| 51 | +web.run_app(app) |
| 52 | + |
| 53 | +``` |
| 54 | + |
| 55 | +## Default dependencies |
| 56 | + |
| 57 | +By default this library provides only two injectables. `web.Request` and `web.Application`. |
| 58 | + |
| 59 | +```python |
| 60 | + |
| 61 | +async def handler(app: web.Application = Depends()): ... |
| 62 | + |
| 63 | +async def handler2(request: web.Request = Depends()): ... |
| 64 | + |
| 65 | +``` |
| 66 | + |
| 67 | +It's super useful, because you can use these dependencies in |
| 68 | +any other dependency. Here's a more complex example of how you can use this library. |
| 69 | + |
| 70 | + |
| 71 | +```python |
| 72 | +from aiohttp_deps import Router, Depends |
| 73 | +from aiohttp import web |
| 74 | + |
| 75 | +router = Router() |
| 76 | + |
| 77 | + |
| 78 | +async def get_db_session(app: web.Application = Depends()): |
| 79 | + async with app["db"] as sess: |
| 80 | + yield sess |
| 81 | + |
| 82 | + |
| 83 | +class MyDAO: |
| 84 | + def __init__(self, session=Depends(get_db_session)): |
| 85 | + self.session = session |
| 86 | + |
| 87 | + async def get_objects(self) -> list[object]: |
| 88 | + return await self.session.execute("SELECT 1") |
| 89 | + |
| 90 | + |
| 91 | +@router.get("/") |
| 92 | +async def handler(db_session: MyDAO = Depends()): |
| 93 | + objs = await db_session.get_objects() |
| 94 | + return web.json_response({"objects": objs}) |
| 95 | + |
| 96 | +``` |
| 97 | + |
| 98 | +If you do something like this, you would never think about initializing your DAO. You can just inject it and that's it. |
| 99 | + |
| 100 | + |
| 101 | +# Built-in dependencies |
| 102 | + |
| 103 | +This library also provides you with some default dependencies that can help you in building the best web-service. |
| 104 | + |
| 105 | +## Json |
| 106 | + |
| 107 | +To parse json, create a pydantic model and add a dependency to your handler. |
| 108 | + |
| 109 | + |
| 110 | +```python |
| 111 | +from aiohttp import web |
| 112 | +from pydantic import BaseModel |
| 113 | +from aiohttp_deps import Router, Json, Depends |
| 114 | + |
| 115 | +router = Router() |
| 116 | + |
| 117 | + |
| 118 | +class UserInfo(BaseModel): |
| 119 | + name: str |
| 120 | + |
| 121 | + |
| 122 | +@router.post("/users") |
| 123 | +async def new_data(user: UserInfo = Depends(Json())): |
| 124 | + return web.json_response({"user": user.dict()}) |
| 125 | + |
| 126 | +``` |
| 127 | + |
| 128 | +This dependency automatically validates data and send |
| 129 | +errors if the data doesn't orrelate with schema or body is not a valid json. |
| 130 | + |
| 131 | +If you want to make this data optional, just mark it as optional. |
| 132 | + |
| 133 | +```python |
| 134 | +@router.post("/users") |
| 135 | +async def new_data(user: Optional[UserInfo] = Depends(Json())): |
| 136 | + if user is None: |
| 137 | + return web.json_response({"user": None}) |
| 138 | + return web.json_response({"user": user.dict()}) |
| 139 | + |
| 140 | +``` |
| 141 | + |
| 142 | +## Headers |
| 143 | + |
| 144 | +You can get and validate headers using `Header` dependency. |
| 145 | + |
| 146 | +Let's try to build simple example for authorization. |
| 147 | + |
| 148 | +```python |
| 149 | +from aiohttp_deps import Router, Header, Depends |
| 150 | +from aiohttp import web |
| 151 | + |
| 152 | +router = Router() |
| 153 | + |
| 154 | + |
| 155 | +def decode_token(authorization: str = Depends(Header())) -> str: |
| 156 | + if authorization == "secret": |
| 157 | + # Let's pretend that here we |
| 158 | + # decode our token. |
| 159 | + return authorization |
| 160 | + raise web.HTTPUnauthorized() |
| 161 | + |
| 162 | + |
| 163 | +@router.get("/secret_data") |
| 164 | +async def new_data(token: str = Depends(decode_token)) -> web.Response: |
| 165 | + return web.json_response({"secret": "not a secret"}) |
| 166 | + |
| 167 | +``` |
| 168 | + |
| 169 | +As you can see, header name to parse is equal to the |
| 170 | +name of a parameter that introduces Header dependency. |
| 171 | + |
| 172 | +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: |
| 173 | + |
| 174 | +```python |
| 175 | +def decode_token(auth: str = Depends(Header(alias="Authorization"))) -> str: |
| 176 | +``` |
| 177 | + |
| 178 | +Headers can also be parsed to types. If you want a header to be parsed as int, just add the typehint. |
| 179 | + |
| 180 | +```python |
| 181 | +def decode_token(meme_id: int = Depends(Header())) -> str: |
| 182 | +``` |
| 183 | + |
| 184 | +If you want to get list of values of one header, use parameter `multiple=True`. |
| 185 | + |
| 186 | +```python |
| 187 | +def decode_token(meme_id: list[int] = Depends(Header(multiple=True))) -> str: |
| 188 | + |
| 189 | +``` |
| 190 | + |
| 191 | +And, of course, you can provide this dependency with default value if the value from user cannot be parsed for some reason. |
| 192 | + |
| 193 | +```python |
| 194 | +def decode_token(meme_id: str = Depends(Header(default="not-a-secret"))) -> str: |
| 195 | +``` |
| 196 | + |
| 197 | + |
| 198 | +# Queries |
| 199 | + |
| 200 | +You can depend on `Query` to get and parse query parameters. |
| 201 | + |
| 202 | +```python |
| 203 | +from aiohttp_deps import Router, Query, Depends |
| 204 | +from aiohttp import web |
| 205 | + |
| 206 | +router = Router() |
| 207 | + |
| 208 | + |
| 209 | +@router.get("/shop") |
| 210 | +async def shop(item_id: str = Depends(Query())) -> web.Response: |
| 211 | + return web.json_response({"id": item_id}) |
| 212 | + |
| 213 | +``` |
| 214 | + |
| 215 | +the name of the parameter is the same as the name of function parameter. |
| 216 | + |
| 217 | +The Query dependency is acually the same as the Header dependency, so everything about the `Header` dependency also applies to `Query`. |
| 218 | + |
| 219 | +## Views |
| 220 | + |
| 221 | +If you use views as handlers, please use View class from `aiohttp_deps`, otherwise the magic won't work. |
| 222 | + |
| 223 | +```python |
| 224 | +from aiohttp_deps import Router, View, Depends |
| 225 | +from aiohttp import web |
| 226 | + |
| 227 | +router = Router() |
| 228 | + |
| 229 | + |
| 230 | +@router.view("/view") |
| 231 | +class MyView(View): |
| 232 | + async def get(self, app: web.Application = Depends()): |
| 233 | + return web.json_response({"app": str(app)}) |
| 234 | + |
| 235 | +``` |
0 commit comments