bind db engine and redis client to app for global usage

[pessimo]

I've been using flask and tornado for some time, and I used to create a DB engine and Redis client and something like the fileserver, then bind them to app.config in the main app.py file.

# in the main app,py file
from sqlalchemy import create_engine
from redis import Redis

dbe = create_engine(...)
app.config['dbe'] = dbe

redis_cli = Redis(...)
app.config['redis_cli'] = redis_cli

then I can from flask import current_app as app and use the app.config['dbe'] to execute sqls.

# in the app view file
from flask import current_app as app
app.config['dbe'].execute(sql, **kwargs)
app.config['redis_cli'].get(...)

With this method, I don't need to initialize the DB engine and Redis client every time, and all the pre-initialized DB engine can be used in the app views(i just treat them as handlers or routers)

I'm wondering is there an equivalent method in fastAPI? Or, I need to create them every time?

[old-syniex]

#1028 I suggest to read the last comment of tigalo

[pessimo]

thanks for the reply, it seems that tiangolo thinks And thread locals are a tool that only works with simple threaded applications, it doesn't work in async contexts. , and he mentioned In those cases (and all the cases in modern Python) you need instead context vars., but how could I create context vars in fastapi?

[leonh]

It is possible to do this and very useful.

Although currently it means making a change to Starlette. Which (according to my needs) incorrectly overwrites the 'app' parameter set in the ASGI context. See Kludex/starlette#977

In my code (which has only been tested in development! ) I'm using app.extra['redis'] to store a redis connection pool.

I have something along the lines of ...

@app.on_event("startup")
async def handle_startup():
    try:
        pool = await aioredis.create_redis_pool(
            (REDIS_HOST, REDIS_PORT), encoding='utf-8')
        logger.info(f"Connected to Redis on {REDIS_HOST} {REDIS_PORT}")
        app.extra['redis'] = pool
    except ConnectionRefusedError as e:
        logger.info(f"cannot connect to redis on {REDIS_HOST} {REDIS_PORT}")
        return

Then in my endpoint handlers i'll have something like...

@router.put("/blah/de/blah/") 
        async def update(request: Request)
            redis = request.app.extra['redis']

It has worked for me so far in my limited testing.... I'd be interested if others can see any problems with this approach?

I tried Contextvars but didnt work for me seems to only make sense to for long lived coroutines for example a websocket connection.

[a137x]

Works the same way for websockets, like:

@router.websocket_route("/ws")
class WSEndpoint(WebSocketEndpoint):
    async def on_connect(self, websocket: WebSocket) -> None:
        await websocket.app.extra['redis'].info()

[pessimo]

I just find that nsidnev made a solution in https://github.com/nsidnev/fastapi-realworld-example-app and the way he/she did it is storing DB connection in app.state.pool, which is actually the same solution in #1028 provided by mikelibg as well as what a137x did above, also leonh used the same solution. But storing DB connection this way might not be recommended, since we didn't get a confirmation in #1028 from tigalo, there might be some profound reasons. Hope someone could explain this in a more detailed way. Thanks in advance.

[buddha314]

I was hoping for a solution similar to using the app_context in Flask, as in this SO: https://stackoverflow.com/questions/37948357/flask-with-celery-application-context-not-available Does anyone know of other workarounds? The problem specifically arises when trying to do tests.

[baby5]

@pessimo you can try this way by using a global redis instance

/project/client.py

redis = None

/project/main.py

import aioredis
import FastAPI

import client

app = FastAPI()


@app.on_event("startup")
async def startup_event():
    client.redis = await aioredis.create_redis_pool("redis://localhost", encoding="utf8")


@app.on_event("shutdown")
async def shutdown_event():
    client.redis.close()
    await client.redis.wait_closed()

then, you can use redis everywhere like:

import client

client.redis.get("key")

[pessimo]

@baby5 thanks for the reply. My question may not be very precise. Actually, what I want is to figure out why Flask style request context is somehow called "black magic", and we don't get a similar implementation in FastAPI. Are there some fatal drawbacks? Or, things get wrong when facing async situations? As I mentioned in the previous reply, there are several solutions from the community, but they're not confirmed by tigalo himself.

I think you provide a good solution. But, I have no idea if this global instance gets handled by the app correctly. Will the app reuse this Redis connection on every request? Or, will it just establish a new connection to handle every request?

[old-syniex]

This is what i am doing on my services:

main.py

import asyncio

import uvloop  # type: ignore

from fastapi import FastAPI, status
from fastapi.exceptions import HTTPException
from pydantic import ValidationError

from . import config
from .events import create_start_app_handler
from .resources import exception_handlers as handlers

asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())


def get_application() -> FastAPI:
    application = FastAPI(
        title=config.APP_TITLE,
        debug=config.API_DEBUG,
        version=config.APP_VERSION,
        docs_url=None,
        redoc_url=None,
        root_path=config.PROXY_PREFIX,
    )
    application.add_event_handler("startup", create_start_app_handler(application))

    application.add_exception_handler(
        status.HTTP_401_UNAUTHORIZED, handlers.http_unauthorized_handler
    )
    application.add_exception_handler(HTTPException, handlers.http_error_handler)
    application.add_exception_handler(ValidationError, handlers.http422_error_handler)
    return application


app = get_application()

#events.py

from typing import Awaitable, Callable

from fastapi import FastAPI

from .redis.events import connect_to_redis


def create_start_app_handler(app: FastAPI) -> Callable[[], Awaitable[None]]:
    async def start_app() -> None:
        await connect_to_redis(app)

    return start_app

#redis/events.py

import aredis  # type: ignore
from fastapi import FastAPI

from .. import config
import logging


async def connect_to_redis(app: FastAPI) -> None:
    logging.info(f"Connecting to {config.REDIS_URL}")
    redis: aredis.StrictRedis = aredis.StrictRedis.from_url(str(config.REDIS_URL))
    app.state.redis = redis
    logging.info("Connection established")

#redis/base.py

from typing import Optional

from aredis import StrictRedis
from ..logging import get_logger
from ..main import app

logger = get_logger("redis")


def get_redis() -> Optional[StrictRedis]:
    if not hasattr(app.state, "redis"):
        logger.error("Redis is not enabled")
        return None
    else:
        return app.state.redis

hope it helps
This way you get get redis from everywhere, same instance.