feat(validation): complete validation editor and master data modules

validation-suite/backend/mercury/app/repositories/base.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+from typing import Generic, TypeVar
+
+from sqlalchemy import Select, func, or_, select
+from sqlalchemy.orm import Session
+
+ModelT = TypeVar("ModelT")
+
+
+class Repository(Generic[ModelT]):
+    model: type[ModelT]
+
+    def __init__(self, session: Session) -> None:
+        self.session = session
+
+    def get(self, item_id: str) -> ModelT | None:
+        return self.session.get(self.model, item_id)
+
+    search_columns: tuple[str, ...] = ()
+
+    def _search_statement(self, search: str | None = None) -> Select[tuple[ModelT]]:
+        statement: Select[tuple[ModelT]] = select(self.model)
+        if search and self.search_columns:
+            term = f"%{search.strip()}%"
+            filters = [getattr(self.model, column).ilike(term) for column in self.search_columns]
+            statement = statement.where(or_(*filters))
+        return statement
+
+    def list(self, limit: int = 20, offset: int = 0, search: str | None = None) -> list[ModelT]:
+        statement = self._search_statement(search).offset(offset).limit(limit)
+        return list(self.session.scalars(statement))
+
+    def count(self, search: str | None = None) -> int:
+        subquery = self._search_statement(search).subquery()
+        return self.session.scalar(select(func.count()).select_from(subquery)) or 0
+
+    def add(self, item: ModelT) -> ModelT:
+        self.session.add(item)
+        self.session.flush()
+        return item
+
+    def delete(self, item: ModelT) -> None:
+        self.session.delete(item)
+        self.session.flush()

validation-suite/backend/mercury/app/repositories/domain.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.models.customer import Customer
+from app.models.device import Device
+from app.models.document import Document
+from app.models.equipment import Equipment
+from app.models.contact import Contact
+from app.models.location import Location
+from app.models.user import User
+from app.models.validation import Validation
+from app.repositories.base import Repository
+
+
+class UserRepository(Repository[User]):
+    model = User
+
+    def by_email(self, email: str) -> User | None:
+        return self.session.scalar(select(User).where(User.email == email.lower()))
+
+
+class CustomerRepository(Repository[Customer]):
+    model = Customer
+    search_columns = ("name", "city", "email", "phone")
+
+
+class LocationRepository(Repository[Location]):
+    model = Location
+    search_columns = ("name", "city", "room")
+
+
+class ContactRepository(Repository[Contact]):
+    model = Contact
+    search_columns = ("full_name", "function", "email", "phone")
+
+
+class DeviceRepository(Repository[Device]):
+    model = Device
+    search_columns = ("manufacturer", "model", "serial_number", "device_type")
+
+
+class EquipmentRepository(Repository[Equipment]):
+    model = Equipment
+    search_columns = ("manufacturer", "model", "serial_number")
+
+
+class ValidationRepository(Repository[Validation]):
+    model = Validation
+
+
+class DocumentRepository(Repository[Document]):
+    model = Document
+
+
+def repositories(session: Session) -> dict[str, Repository]:
+    return {
+        "users": UserRepository(session),
+        "customers": CustomerRepository(session),
+        "locations": LocationRepository(session),
+        "contacts": ContactRepository(session),
+        "devices": DeviceRepository(session),
+        "equipment": EquipmentRepository(session),
+        "validations": ValidationRepository(session),
+        "documents": DocumentRepository(session),
+    }