generated from coulomb/repo-seed
37 lines
836 B
Python
37 lines
836 B
Python
from collections.abc import AsyncIterator
|
|
from functools import lru_cache
|
|
|
|
from sqlalchemy.ext.asyncio import (
|
|
AsyncEngine,
|
|
AsyncSession,
|
|
async_sessionmaker,
|
|
create_async_engine,
|
|
)
|
|
from sqlalchemy.orm import DeclarativeBase
|
|
|
|
from .config import get_settings
|
|
|
|
|
|
class Base(DeclarativeBase):
|
|
pass
|
|
|
|
|
|
def make_engine(database_url: str) -> AsyncEngine:
|
|
return create_async_engine(database_url, pool_pre_ping=True)
|
|
|
|
|
|
@lru_cache
|
|
def get_engine() -> AsyncEngine:
|
|
return make_engine(get_settings().database_url)
|
|
|
|
|
|
@lru_cache
|
|
def get_sessionmaker() -> async_sessionmaker[AsyncSession]:
|
|
return async_sessionmaker(get_engine(), expire_on_commit=False)
|
|
|
|
|
|
async def session_scope() -> AsyncIterator[AsyncSession]:
|
|
async_session = get_sessionmaker()
|
|
async with async_session() as session:
|
|
yield session
|