db

app/db.py

@@ -0,0 +1,40 @@
+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()()