app/db.py

"""SQLite connection + transaction helpers.

Every connection enables foreign keys. ``with_tx`` gives an atomic unit of work:
the wrapped callable shares one connection/transaction, so a raised exception
(e.g. CycleError) rolls the whole patch back.
"""

import sqlite3
from collections.abc import Iterator
from contextlib import contextmanager

from app import config

_db_path = config.DB_PATH


def set_db_path(path: str) -> None:
    """Point the engine at a different SQLite file (used by tests)."""
    global _db_path
    _db_path = path


def get_db_path() -> str:
    return _db_path


def connect() -> sqlite3.Connection:
    conn = sqlite3.connect(_db_path)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    return conn


@contextmanager
def with_tx() -> Iterator[sqlite3.Connection]:
    """Atomic transaction. Commits on success, rolls back on any exception."""
    conn = connect()
    try:
        conn.execute("BEGIN")
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()