29 lines
728 B
Python
29 lines
728 B
Python
|
from collections.abc import Generator
|
||
|
|
|
||
|
|
from sqlalchemy import create_engine
|
||
|
|
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
|
||
|
|
|
||
|
|
from app.config import get_settings
|
||
|
|
|
||
|
|
|
||
|
|
class PooBase(DeclarativeBase):
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
settings = get_settings()
|
||
|
|
|
||
|
|
connect_args: dict[str, object] = {}
|
||
|
|
if settings.poo_database_url.startswith("sqlite"):
|
||
|
|
connect_args["check_same_thread"] = False
|
||
|
|
|
||
|
|
poo_engine = create_engine(settings.poo_database_url, connect_args=connect_args)
|
||
|
|
PooSessionLocal = sessionmaker(bind=poo_engine, autoflush=False, autocommit=False, class_=Session)
|
||
|
|
|
||
|
|
|
||
|
|
def get_poo_db_session() -> Generator[Session, None, None]:
|
||
|
|
session = PooSessionLocal()
|
||
|
|
try:
|
||
|
|
yield session
|
||
|
|
finally:
|
||
|
|
session.close()
|