feat(db): module repository with crud and filtered list
This commit is contained in:
DumiFlex
@@ -0,0 +1,164 @@
|
||||
"""Repository classes wrapping SQL access to library tables.
|
||||
|
||||
Repositories take a ``sqlite3.Connection``. Each public method opens a
|
||||
transaction via ``with conn:`` so callers don't need explicit commit
|
||||
calls. Rows are returned as plain dicts (sqlite3.Row converted).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as _dt
|
||||
import json
|
||||
import re
|
||||
import secrets
|
||||
import sqlite3
|
||||
from typing import Any
|
||||
|
||||
_TYPE_PREFIX = {"wildcard": "wc", "fixed_values": "fv"}
|
||||
_SLUG_RE = re.compile(r"[^a-z0-9]+")
|
||||
|
||||
|
||||
class ModuleNotFound(LookupError):
|
||||
"""Raised when a requested module id does not exist."""
|
||||
|
||||
|
||||
def _slug(name: str) -> str:
|
||||
s = _SLUG_RE.sub("_", name.lower()).strip("_")
|
||||
return s[:24] or "module"
|
||||
|
||||
|
||||
def _now() -> str:
|
||||
return _dt.datetime.now(_dt.UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
|
||||
|
||||
def _row_to_module(row: sqlite3.Row) -> dict[str, Any]:
|
||||
return {
|
||||
"id": row["id"],
|
||||
"type": row["type"],
|
||||
"name": row["name"],
|
||||
"description": row["description"],
|
||||
"category_id": row["category_id"],
|
||||
"tags": json.loads(row["tags"]),
|
||||
"is_favorite": bool(row["is_favorite"]),
|
||||
"payload": json.loads(row["payload"]),
|
||||
"version": row["version"],
|
||||
"created_at": row["created_at"],
|
||||
"updated_at": row["updated_at"],
|
||||
}
|
||||
|
||||
|
||||
class ModuleRepository:
|
||||
def __init__(self, conn: sqlite3.Connection) -> None:
|
||||
self._conn = conn
|
||||
|
||||
def _gen_id(self, type: str, name: str) -> str:
|
||||
prefix = _TYPE_PREFIX.get(type, "mod")
|
||||
return f"{prefix}_{_slug(name)}_{secrets.token_hex(2)}"
|
||||
|
||||
def create(
|
||||
self,
|
||||
*,
|
||||
type: str,
|
||||
name: str,
|
||||
description: str,
|
||||
category_id: str | None,
|
||||
tags: list[str],
|
||||
payload: dict[str, Any],
|
||||
is_favorite: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
mid = self._gen_id(type, name)
|
||||
now = _now()
|
||||
with self._conn:
|
||||
self._conn.execute(
|
||||
"INSERT INTO modules("
|
||||
"id, type, name, description, category_id, tags, "
|
||||
"is_favorite, payload, version, created_at, updated_at"
|
||||
") VALUES(?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?);",
|
||||
(
|
||||
mid, type, name, description, category_id,
|
||||
json.dumps(tags), int(is_favorite),
|
||||
json.dumps(payload), now, now,
|
||||
),
|
||||
)
|
||||
return self.get(mid)
|
||||
|
||||
def get(self, module_id: str) -> dict[str, Any]:
|
||||
row = self._conn.execute(
|
||||
"SELECT * FROM modules WHERE id = ?;", (module_id,),
|
||||
).fetchone()
|
||||
if row is None:
|
||||
raise ModuleNotFound(module_id)
|
||||
return _row_to_module(row)
|
||||
|
||||
def update(
|
||||
self,
|
||||
module_id: str,
|
||||
*,
|
||||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
category_id: str | None | object = ...,
|
||||
tags: list[str] | None = None,
|
||||
payload: dict[str, Any] | None = None,
|
||||
is_favorite: bool | None = None,
|
||||
) -> dict[str, Any]:
|
||||
existing = self.get(module_id)
|
||||
new = {
|
||||
"name": existing["name"] if name is None else name,
|
||||
"description": existing["description"] if description is None else description,
|
||||
"category_id": existing["category_id"] if category_id is ... else category_id,
|
||||
"tags": existing["tags"] if tags is None else tags,
|
||||
"payload": existing["payload"] if payload is None else payload,
|
||||
"is_favorite": existing["is_favorite"] if is_favorite is None else is_favorite,
|
||||
}
|
||||
now = _now()
|
||||
with self._conn:
|
||||
self._conn.execute(
|
||||
"UPDATE modules SET "
|
||||
"name = ?, description = ?, category_id = ?, tags = ?, "
|
||||
"is_favorite = ?, payload = ?, version = version + 1, "
|
||||
"updated_at = ? "
|
||||
"WHERE id = ?;",
|
||||
(
|
||||
new["name"], new["description"], new["category_id"],
|
||||
json.dumps(new["tags"]), int(new["is_favorite"]),
|
||||
json.dumps(new["payload"]), now, module_id,
|
||||
),
|
||||
)
|
||||
return self.get(module_id)
|
||||
|
||||
def delete(self, module_id: str) -> None:
|
||||
self.get(module_id) # raises ModuleNotFound if absent
|
||||
with self._conn:
|
||||
self._conn.execute("DELETE FROM modules WHERE id = ?;", (module_id,))
|
||||
|
||||
def list(
|
||||
self,
|
||||
*,
|
||||
type: str | None = None,
|
||||
category_id: str | None = None,
|
||||
query: str | None = None,
|
||||
favorites_only: bool = False,
|
||||
limit: int | None = None,
|
||||
offset: int = 0,
|
||||
) -> list[dict[str, Any]]:
|
||||
clauses: list[str] = []
|
||||
params: list[Any] = []
|
||||
if type is not None:
|
||||
clauses.append("type = ?")
|
||||
params.append(type)
|
||||
if category_id is not None:
|
||||
clauses.append("category_id = ?")
|
||||
params.append(category_id)
|
||||
if query:
|
||||
clauses.append("name LIKE ? COLLATE NOCASE")
|
||||
params.append(f"%{query}%")
|
||||
if favorites_only:
|
||||
clauses.append("is_favorite = 1")
|
||||
sql = "SELECT * FROM modules"
|
||||
if clauses:
|
||||
sql += " WHERE " + " AND ".join(clauses)
|
||||
sql += " ORDER BY updated_at DESC, id DESC"
|
||||
if limit is not None:
|
||||
sql += " LIMIT ? OFFSET ?"
|
||||
params.extend([limit, offset])
|
||||
rows = self._conn.execute(sql, params).fetchall()
|
||||
return [_row_to_module(r) for r in rows]
|
||||
@@ -0,0 +1,163 @@
|
||||
"""Tests for ModuleRepository CRUD + filtering."""
|
||||
import pytest
|
||||
|
||||
from engine.db.repositories import ModuleNotFound, ModuleRepository
|
||||
|
||||
|
||||
def _new_payload() -> dict:
|
||||
return {"options": [{"id": "opt_1", "value": "red", "weight": 1}]}
|
||||
|
||||
|
||||
def test_create_assigns_id_and_timestamps(wp_db):
|
||||
repo = ModuleRepository(wp_db)
|
||||
row = repo.create(
|
||||
type="wildcard", name="colors", description="",
|
||||
category_id=None, tags=["a", "b"], payload=_new_payload(),
|
||||
)
|
||||
assert row["id"].startswith("wc_colors_")
|
||||
assert row["version"] == 1
|
||||
assert row["created_at"]
|
||||
assert row["updated_at"] == row["created_at"]
|
||||
assert row["tags"] == ["a", "b"]
|
||||
assert row["payload"] == _new_payload()
|
||||
|
||||
|
||||
def test_get_returns_existing_row(wp_db):
|
||||
repo = ModuleRepository(wp_db)
|
||||
created = repo.create(
|
||||
type="fixed_values", name="lens", description="",
|
||||
category_id=None, tags=[], payload={"values": []},
|
||||
)
|
||||
got = repo.get(created["id"])
|
||||
assert got["id"] == created["id"]
|
||||
assert got["type"] == "fixed_values"
|
||||
|
||||
|
||||
def test_get_raises_module_not_found(wp_db):
|
||||
repo = ModuleRepository(wp_db)
|
||||
with pytest.raises(ModuleNotFound):
|
||||
repo.get("does_not_exist")
|
||||
|
||||
|
||||
def test_update_bumps_version_and_updated_at(wp_db):
|
||||
import time
|
||||
repo = ModuleRepository(wp_db)
|
||||
created = repo.create(
|
||||
type="wildcard", name="colors", description="",
|
||||
category_id=None, tags=[], payload=_new_payload(),
|
||||
)
|
||||
time.sleep(1.05) # ensure ISO-second-precision updated_at differs
|
||||
updated = repo.update(created["id"], name="palette", payload=_new_payload())
|
||||
assert updated["name"] == "palette"
|
||||
assert updated["version"] == 2
|
||||
assert updated["updated_at"] >= created["updated_at"]
|
||||
|
||||
|
||||
def test_delete_removes_row(wp_db):
|
||||
repo = ModuleRepository(wp_db)
|
||||
created = repo.create(
|
||||
type="wildcard", name="x", description="",
|
||||
category_id=None, tags=[], payload=_new_payload(),
|
||||
)
|
||||
repo.delete(created["id"])
|
||||
with pytest.raises(ModuleNotFound):
|
||||
repo.get(created["id"])
|
||||
|
||||
|
||||
def test_delete_missing_raises(wp_db):
|
||||
repo = ModuleRepository(wp_db)
|
||||
with pytest.raises(ModuleNotFound):
|
||||
repo.delete("ghost")
|
||||
|
||||
|
||||
def test_list_filters_by_type(wp_db):
|
||||
repo = ModuleRepository(wp_db)
|
||||
repo.create(type="wildcard", name="a", description="",
|
||||
category_id=None, tags=[], payload=_new_payload())
|
||||
repo.create(type="fixed_values", name="b", description="",
|
||||
category_id=None, tags=[], payload={"values": []})
|
||||
wcs = repo.list(type="wildcard")
|
||||
assert len(wcs) == 1
|
||||
assert wcs[0]["type"] == "wildcard"
|
||||
|
||||
|
||||
def test_list_filters_by_category(wp_db):
|
||||
wp_db.execute(
|
||||
"INSERT INTO module_categories(id, name) VALUES('style', 'Style');"
|
||||
)
|
||||
wp_db.commit()
|
||||
repo = ModuleRepository(wp_db)
|
||||
repo.create(type="wildcard", name="a", description="",
|
||||
category_id="style", tags=[], payload=_new_payload())
|
||||
repo.create(type="wildcard", name="b", description="",
|
||||
category_id=None, tags=[], payload=_new_payload())
|
||||
rows = repo.list(category_id="style")
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["name"] == "a"
|
||||
|
||||
|
||||
def test_list_search_matches_name_case_insensitive(wp_db):
|
||||
repo = ModuleRepository(wp_db)
|
||||
repo.create(type="wildcard", name="ColorPalette", description="",
|
||||
category_id=None, tags=[], payload=_new_payload())
|
||||
repo.create(type="wildcard", name="lighting", description="",
|
||||
category_id=None, tags=[], payload=_new_payload())
|
||||
rows = repo.list(query="color")
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["name"] == "ColorPalette"
|
||||
|
||||
|
||||
def test_list_orders_by_updated_at_desc(wp_db):
|
||||
import time
|
||||
repo = ModuleRepository(wp_db)
|
||||
a = repo.create(type="wildcard", name="a", description="",
|
||||
category_id=None, tags=[], payload=_new_payload())
|
||||
time.sleep(1.05) # ISO-second-precision: ensure b > a
|
||||
b = repo.create(type="wildcard", name="b", description="",
|
||||
category_id=None, tags=[], payload=_new_payload())
|
||||
rows = repo.list()
|
||||
assert [r["id"] for r in rows] == [b["id"], a["id"]]
|
||||
|
||||
|
||||
def test_list_favorites_only(wp_db):
|
||||
repo = ModuleRepository(wp_db)
|
||||
repo.create(type="wildcard", name="a", description="",
|
||||
category_id=None, tags=[], payload=_new_payload(), is_favorite=True)
|
||||
repo.create(type="wildcard", name="b", description="",
|
||||
category_id=None, tags=[], payload=_new_payload())
|
||||
rows = repo.list(favorites_only=True)
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["name"] == "a"
|
||||
|
||||
|
||||
def test_list_limit_offset(wp_db):
|
||||
import time
|
||||
repo = ModuleRepository(wp_db)
|
||||
for n in ["a", "b", "c"]:
|
||||
repo.create(type="wildcard", name=n, description="",
|
||||
category_id=None, tags=[], payload=_new_payload())
|
||||
time.sleep(1.05)
|
||||
rows = repo.list(limit=1, offset=1)
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["name"] == "b"
|
||||
|
||||
|
||||
def test_update_keeps_unchanged_fields(wp_db):
|
||||
repo = ModuleRepository(wp_db)
|
||||
created = repo.create(
|
||||
type="wildcard", name="x", description="hello",
|
||||
category_id=None, tags=["t1"], payload=_new_payload(),
|
||||
)
|
||||
updated = repo.update(created["id"], name="y") # only name changes
|
||||
assert updated["description"] == "hello"
|
||||
assert updated["tags"] == ["t1"]
|
||||
assert updated["payload"] == _new_payload()
|
||||
|
||||
|
||||
def test_create_id_uses_fv_prefix_for_fixed_values(wp_db):
|
||||
repo = ModuleRepository(wp_db)
|
||||
row = repo.create(
|
||||
type="fixed_values", name="lens", description="",
|
||||
category_id=None, tags=[], payload={"values": []},
|
||||
)
|
||||
assert row["id"].startswith("fv_lens_")
|
||||
Reference in New Issue
Block a user