Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- #session_factory.py
- class SessionFactory:
- def __init__(self, postgres_url: str):
- self.engine: AsyncEngine = create_async_engine(postgres_url, echo=False)
- self.async_session_maker: async_sessionmaker[AsyncSession] = async_sessionmaker(
- self.engine,
- class_=AsyncSession,
- expire_on_commit=False,
- )
- async def dispose(self) -> None:
- await self.engine.dispose()
- async def __call__(self) -> AsyncGenerator[AsyncSession, None]:
- async with self.async_session_maker() as session:
- yield session
- #database.py
- session_factory: SessionFactory | None = None
- async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
- if session_factory is None:
- raise RuntimeError("SessionFactory is not initialized")
- async for session in session_factory():
- yield session
- async def get_db_stub():
- raise NotImplementedError
- get_async_session_context = contextlib.asynccontextmanager(get_async_session)
- #routers.py
- @router.post(
- "/", response_model=ReleaseCreateResponse, status_code=status.HTTP_201_CREATED
- )
- async def create_release(
- user: Annotated[User, Depends(current_active_user)],
- session: Annotated[AsyncSession, Depends(get_db_stub)],
- ):
- ...
- #main.py
- @asynccontextmanager
- async def lifespan(app: FastAPI):
- db.session_factory = SessionFactory(settings.db.url_postgres)
- app.dependency_overrides[AsyncSession] = db.get_async_session
- print(app.dependency_overrides[AsyncSession])
- >> function get_async_session at 0x7fd44df4e2a0
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement