How to initialize a database connection only once when deploying services using FastAPI

[TheHonestBob]

Describe the bug

The simplest way is to define conn and cursor in the initialization method, but it fails because the init method cannot be asynchronous

To Reproduce

无

Expected behavior

I only want to connect to the database once for a FastAPI service, rather than having to connect to the database every time a request is made

Logs/tracebacks

无

Python Version

$ python --version
python = 3.9

aiomysql Version

$ python -m pip show aiomysql
0.2.0

PyMySQL Version

$ python -m pip show PyMySQL
1.1.0

SQLAlchemy Version

$ python -m pip show sqlalchemy

OS

linux

Database type and version

SELECT VERSION();
无

Additional context

No response

Code of Conduct

• I agree to follow the aio-libs Code of Conduct

[yarodevuci]

i am curious too

[asparagusbeef]

You can use a lifespan, something like

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.db_engine = await create_my_async_engine()
    yield

And then you can define some function

def get_engine(request: Request) -> sqlalchemy.ext.asyncio.AsyncEngine:
    return request.app.state.db_engine

And in your routes you can access your engine like

@userRouter.get("/{user_id}")
async def get_user(user_id: int, engine: sqlalchemy.ext.asyncio.AsyncEngine = Depends(get_engine))
    async with engine.begin() as conn:
        result = await conn.execute(users.select().where(users.c.id == user_id))
        return result.first()