python-test/Module 5/db/session.py

10 lines
No EOL
375 B
Python

from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
engine = create_async_engine("sqlite+aiosqlite:///database.db", echo=True)
SessionAsync = async_sessionmaker(autoflush=True, bind=engine)
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with SessionAsync() as db:
yield db