2025-02-17 19:44:17 +05:30

21 lines
516 B
Python

from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession
from starlette.requests import Request
async def get_db_session(request: Request) -> AsyncGenerator[AsyncSession, None]:
"""
Create and get database session.
:param request: current request.
:yield: database session.
"""
session: AsyncSession = request.app.state.db_session_factory()
try: # noqa: WPS501
yield session
await session.commit()
finally:
await session.close()