mirror of
https://github.com/FutureOfMedTech-FITM-hack/backend.git
synced 2024-11-26 00:03:44 +03:00
27 lines
732 B
Python
27 lines
732 B
Python
|
from typing import AsyncGenerator
|
||
|
|
|
||
|
|
from redis.asyncio import Redis
|
||
|
|
from starlette.requests import Request
|
||
|
|
|
||
|
|
|
||
|
|
async def get_redis_pool(
|
||
|
|
request: Request,
|
||
|
|
) -> AsyncGenerator[Redis, None]: # pragma: no cover
|
||
|
|
"""
|
||
|
|
Returns connection pool.
|
||
|
|
|
||
|
|
You can use it like this:
|
||
|
|
|
||
|
|
>>> from redis.asyncio import ConnectionPool, Redis
|
||
|
|
>>>
|
||
|
|
>>> async def handler(redis_pool: ConnectionPool = Depends(get_redis_pool)):
|
||
|
|
>>> async with Redis(connection_pool=redis_pool) as redis:
|
||
|
|
>>> await redis.get('key')
|
||
|
|
|
||
|
|
I use pools so you don't acquire connection till the end of the handler.
|
||
|
|
|
||
|
|
:param request: current request.
|
||
|
|
:returns: redis connection pool.
|
||
|
|
"""
|
||
|
|
return request.app.state.redis_pool
|