antragstracker/backend/src/tracker/db/session.py

28 lines
766 B
Python
Raw Normal View History

"""SQLite database connection management."""
import os
import sqlite3
from contextlib import contextmanager
from pathlib import Path
# Use environment variable or default to relative path
DB_PATH = Path(os.environ.get("DATABASE_PATH", Path(__file__).resolve().parents[4] / "data" / "tracker.db"))
def get_connection(db_path: Path | str | None = None) -> sqlite3.Connection:
path = str(db_path or DB_PATH)
conn = sqlite3.connect(path, detect_types=0)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode = WAL")
conn.execute("PRAGMA foreign_keys = ON")
return conn
@contextmanager
def get_db(db_path: Path | str | None = None):
conn = get_connection(db_path)
try:
yield conn
finally:
conn.close()