33 lines
784 B
Python
33 lines
784 B
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Generator
|
|
from functools import lru_cache
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
|
|
|
|
from gitea_codex_bot.config import get_settings
|
|
|
|
|
|
class Base(DeclarativeBase):
|
|
pass
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def get_engine():
|
|
settings = get_settings()
|
|
return create_engine(settings.sqlalchemy_url, pool_pre_ping=True, future=True)
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def get_session_factory():
|
|
return sessionmaker(bind=get_engine(), class_=Session, autoflush=False, autocommit=False, expire_on_commit=False)
|
|
|
|
|
|
def get_session() -> Generator[Session, None, None]:
|
|
session = get_session_factory()()
|
|
try:
|
|
yield session
|
|
finally:
|
|
session.close()
|