109 lines
4.3 KiB
Python
109 lines
4.3 KiB
Python
# backend/db/base_repository.py
|
||
from __future__ import annotations
|
||
|
||
from typing import TypeVar, Generic, List, Sequence, Any, Mapping, Tuple
|
||
from backend.db.connection import DatabaseConnection
|
||
|
||
# Generic type for the model (your dataclasses such as Member, Service, …)
|
||
T = TypeVar("T")
|
||
|
||
|
||
class BaseRepository(Generic[T]):
|
||
"""
|
||
Very small generic repository that knows how to:
|
||
|
||
* INSERT a dataclass‑like object (any object that implements ``to_dict`` and
|
||
``from_row``)
|
||
* SELECT all rows from a table and turn them into model instances
|
||
* (optionally) UPDATE or DELETE rows – stubs are provided for future use
|
||
"""
|
||
|
||
def __init__(self, db: DatabaseConnection):
|
||
self.db = db
|
||
|
||
# ------------------------------------------------------------------
|
||
# INSERT
|
||
# ------------------------------------------------------------------
|
||
def _insert(self, table: str, obj: T, pk_name: str) -> T:
|
||
"""
|
||
Insert ``obj`` into ``table`` and populate the autoincrement primary‑key
|
||
field named ``pk_name`` on the original object.
|
||
|
||
The model **must** implement:
|
||
* ``to_dict() -> Mapping[str, Any]`` – returns a mapping of column →
|
||
value (including the PK, which we drop here)
|
||
* ``from_row(row: sqlite3.Row) -> Model`` – classmethod used by
|
||
``_select_all``.
|
||
"""
|
||
# 1️⃣ Turn the model into a plain dict and drop the PK column.
|
||
data: Mapping[str, Any] = obj.to_dict() # type: ignore[attr-defined]
|
||
if pk_name not in data:
|
||
raise ValueError(f"Primary‑key column '{pk_name}' not found in model data.")
|
||
# Remove the autoincrement column – SQLite will fill it in.
|
||
data_without_pk = {k: v for k, v in data.items() if k != pk_name}
|
||
|
||
# 2️⃣ Build the column list and matching placeholders.
|
||
columns = ", ".join(data_without_pk.keys())
|
||
placeholders = ", ".join("?" for _ in data_without_pk)
|
||
|
||
sql = f"INSERT INTO {table} ({columns}) VALUES ({placeholders})"
|
||
|
||
# 3️⃣ Execute the statement with a *tuple* of values that matches the
|
||
# number of placeholders.
|
||
cursor = self.db.execute(sql, tuple(data_without_pk.values()))
|
||
|
||
# 4️⃣ SQLite gives us the newly generated row‑id on the cursor.
|
||
setattr(obj, pk_name, cursor.lastrowid) # type: ignore[attr-defined]
|
||
|
||
return obj
|
||
|
||
# ------------------------------------------------------------------
|
||
# SELECT ALL
|
||
# ------------------------------------------------------------------
|
||
def _select_all(self, table: str, model_cls: type[T]) -> List[T]:
|
||
"""
|
||
Return every row from ``table`` as a list of ``model_cls`` instances.
|
||
``model_cls`` must provide a ``from_row`` classmethod that accepts a
|
||
``sqlite3.Row`` and returns an instantiated model.
|
||
"""
|
||
rows = self.db.fetchall(f"SELECT * FROM {table}")
|
||
return [model_cls.from_row(r) for r in rows] # type: ignore[attr-defined]
|
||
|
||
# ------------------------------------------------------------------
|
||
# OPTIONAL UPDATE helper (you can call it from concrete repos)
|
||
# ------------------------------------------------------------------
|
||
def _update(
|
||
self,
|
||
table: str,
|
||
pk_name: str,
|
||
pk_value: Any,
|
||
updates: Mapping[str, Any],
|
||
) -> None:
|
||
"""
|
||
Simple UPDATE helper.
|
||
|
||
Example:
|
||
repo._update(
|
||
table="Members",
|
||
pk_name="MemberId",
|
||
pk_value=42,
|
||
updates={"IsActive": 0, "Notes": "temporarily disabled"},
|
||
)
|
||
"""
|
||
if not updates:
|
||
return # nothing to do
|
||
|
||
set_clause = ", ".join(f"{col}=?" for col in updates)
|
||
sql = f"UPDATE {table} SET {set_clause} WHERE {pk_name} = ?"
|
||
params: Tuple[Any, ...] = tuple(updates.values()) + (pk_value,)
|
||
self.db.execute(sql, params)
|
||
|
||
# ------------------------------------------------------------------
|
||
# OPTIONAL DELETE helper
|
||
# ------------------------------------------------------------------
|
||
def _delete(self, table: str, pk_name: str, pk_value: Any) -> None:
|
||
"""
|
||
Delete a row by primary key.
|
||
"""
|
||
sql = f"DELETE FROM {table} WHERE {pk_name} = ?"
|
||
self.db.execute(sql, (pk_value,)) |