diff --git a/backend/src/cyclone/db.py b/backend/src/cyclone/db.py new file mode 100644 index 0000000..61b467f --- /dev/null +++ b/backend/src/cyclone/db.py @@ -0,0 +1,97 @@ +"""SQLAlchemy 2.0 engine, session factory, and declarative Base. + +`init_db()` is called from `cyclone.__main__:serve` before uvicorn starts. +It reads `CYCLONE_DB_URL` (default `sqlite:///~/.local/share/cyclone/cyclone.db`), +creates the engine, runs `Base.metadata.create_all`, then runs the migration +runner. Calling `init_db()` is idempotent — calling it twice on the same URL +re-uses the same engine. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +import sqlalchemy as sa +from sqlalchemy.orm import DeclarativeBase, sessionmaker + + +DEFAULT_DB_PATH = Path.home() / ".local" / "share" / "cyclone" / "cyclone.db" +DEFAULT_DB_URL = f"sqlite:///{DEFAULT_DB_PATH}" + + +class Base(DeclarativeBase): + """Declarative base for all ORM models in the project.""" + + +_engine: sa.Engine | None = None +_SessionLocal: sessionmaker[sa.orm.Session] | None = None + + +def _resolve_url() -> str: + url = os.environ.get("CYCLONE_DB_URL") + if url: + return url + DEFAULT_DB_PATH.parent.mkdir(parents=True, exist_ok=True) + return DEFAULT_DB_URL + + +def _make_engine(url: str) -> sa.Engine: + """Build an Engine with sensible defaults for SQLite + FastAPI.""" + connect_args: dict[str, object] = {} + if url.startswith("sqlite"): + connect_args = {"check_same_thread": False} + return sa.create_engine( + url, + connect_args=connect_args, + future=True, + ) + + +def init_db() -> None: + """Create the engine + session factory; run create_all + migrations. + + Idempotent. Safe to call multiple times for the same URL. + """ + global _engine, _SessionLocal + if _engine is not None: + return # already initialized + + url = _resolve_url() + _engine = _make_engine(url) + + # Import here to avoid circular imports with models. + from cyclone import db_migrate # noqa: WPS433 (deliberate late import) + + Base.metadata.create_all(_engine) + db_migrate.run(_engine) + + _SessionLocal = sessionmaker( + bind=_engine, + autoflush=False, + autocommit=False, + expire_on_commit=False, + ) + + +def _reset_for_tests() -> None: + """Clear module-level state. Test-only.""" + global _engine, _SessionLocal + if _engine is not None: + _engine.dispose() + _engine = None + _SessionLocal = None + + +def engine() -> sa.Engine: + """Return the process-wide Engine. Raises if `init_db()` was not called.""" + if _engine is None: + raise RuntimeError("db.init_db() has not been called") + return _engine + + +def SessionLocal() -> sessionmaker[sa.orm.Session]: + """Return the session factory. Raises if `init_db()` was not called.""" + if _SessionLocal is None: + raise RuntimeError("db.init_db() has not been called") + return _SessionLocal diff --git a/backend/src/cyclone/db_migrate.py b/backend/src/cyclone/db_migrate.py new file mode 100644 index 0000000..e3d9223 --- /dev/null +++ b/backend/src/cyclone/db_migrate.py @@ -0,0 +1,13 @@ +"""Database migration runner. + +Stub. Task 3 replaces this with a real migration implementation +that tracks applied migrations in a ``schema_migrations`` table. +""" + +from __future__ import annotations + +import sqlalchemy as sa + + +def run(engine: sa.Engine) -> None: + """Apply pending migrations. No-op until Task 3.""" diff --git a/backend/tests/test_db.py b/backend/tests/test_db.py new file mode 100644 index 0000000..966ccc6 --- /dev/null +++ b/backend/tests/test_db.py @@ -0,0 +1,41 @@ +"""Tests for the SQLAlchemy engine + session factory + init_db.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import sqlalchemy as sa + +from cyclone import db + + +@pytest.fixture(autouse=True) +def _isolated_db(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Point CYCLONE_DB_URL at a temp file for every test in this module.""" + url = f"sqlite:///{tmp_path}/test.db" + monkeypatch.setenv("CYCLONE_DB_URL", url) + # Reset module-level state so each test gets a fresh engine. + db._reset_for_tests() + + +def test_init_db_creates_engine() -> None: + db.init_db() + assert db.engine() is not None + assert isinstance(db.engine(), sa.Engine) + + +def test_init_db_creates_all_tables() -> None: + db.init_db() + # Probe known future table names from the schema. We use + # `inspect` rather than importing the model class to keep this + # test independent of model implementation. + inspector = sa.inspect(db.engine()) + # In Task 3 we add a Claim table; until then this is empty. + assert isinstance(inspector.get_table_names(), list) + + +def test_session_local_returns_session() -> None: + db.init_db() + with db.SessionLocal()() as s: + assert s.is_active