41 lines
988 B
Python
41 lines
988 B
Python
from __future__ import annotations
|
|
|
|
from functools import lru_cache
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from app.config import DATA_DIR, get_settings
|
|
from app.db_models import Base
|
|
|
|
|
|
def _sqlite_connect_args(database_url: str) -> dict[str, bool]:
|
|
if database_url.startswith("sqlite"):
|
|
return {"check_same_thread": False}
|
|
return {}
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def get_engine():
|
|
settings = get_settings()
|
|
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
return create_engine(
|
|
settings.database_url,
|
|
echo=settings.database_echo,
|
|
future=True,
|
|
connect_args=_sqlite_connect_args(settings.database_url),
|
|
)
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def get_session_factory():
|
|
return sessionmaker(bind=get_engine(), autoflush=False, autocommit=False, future=True)
|
|
|
|
|
|
def init_db() -> None:
|
|
Base.metadata.create_all(bind=get_engine())
|
|
|
|
|
|
def get_session() -> Session:
|
|
return get_session_factory()()
|