feat(dependencies): update database engine dependencies to support async operations

This commit is contained in:
2026-07-24 11:38:51 -04:00
parent 9f126fedc7
commit e510c94b95
5 changed files with 46 additions and 19 deletions
+14 -1
View File
@@ -5,13 +5,24 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Annotated
from fastapi import Depends, Request
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.engine import Engine
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
from sqlalchemy.orm import Session
if TYPE_CHECKING:
from collections.abc import AsyncIterator, Iterator
def get_engine(request: Request) -> Engine:
"""Get a synchronous database engine from app state."""
return request.app.state.engine
def get_async_engine(request: Request) -> AsyncEngine:
"""Get an asynchronous database engine from app state."""
return request.app.state.engine
def get_db(request: Request) -> Iterator[Session]:
"""Get database session from app state."""
with Session(request.app.state.engine) as session:
@@ -28,5 +39,7 @@ async def get_async_db(request: Request) -> AsyncIterator[AsyncSession]:
yield session
AppEngine = Annotated[Engine, Depends(get_engine)]
AppAsyncEngine = Annotated[AsyncEngine, Depends(get_async_engine)]
DbSession = Annotated[Session, Depends(get_db)]
AsyncDbSession = Annotated[AsyncSession, Depends(get_async_db)]