105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
# myapp/repositories/service.py
|
||
# ------------------------------------------------------------
|
||
# Persistence layer for Service‑related models.
|
||
# ------------------------------------------------------------
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import date, datetime
|
||
from typing import List, Optional, Sequence, Any
|
||
|
||
from backend.db import BaseRepository
|
||
from backend.models import Service as ServiceModel
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# ServiceRepository – handles the ``Services`` table
|
||
# ----------------------------------------------------------------------
|
||
class ServiceRepository(BaseRepository[ServiceModel]):
|
||
"""
|
||
CRUD + query helpers for the ``Services`` table.
|
||
Business rules (e.g. “do not schedule past services”) belong in a
|
||
service layer that composes this repository with the others.
|
||
"""
|
||
|
||
_TABLE = "Services"
|
||
_PK = "ServiceId"
|
||
|
||
# ------------------------------
|
||
# Basic CRUD
|
||
# ------------------------------
|
||
def create(
|
||
self,
|
||
service_type_id: int,
|
||
service_date: date,
|
||
) -> ServiceModel:
|
||
"""
|
||
Insert a new service row.
|
||
|
||
``service_date`` can be a ``datetime.date`` or an ISO‑8601 string.
|
||
"""
|
||
svc = ServiceModel(
|
||
ServiceId=-1, # placeholder – will be overwritten
|
||
ServiceTypeId=service_type_id,
|
||
ServiceDate=service_date,
|
||
)
|
||
return self._insert(self._TABLE, svc, self._PK)
|
||
|
||
def get_by_id(self, service_id: int) -> Optional[ServiceModel]:
|
||
sql = f"SELECT * FROM {self._TABLE} WHERE {self._PK} = ?"
|
||
row = self.db.fetchone(sql, (service_id,))
|
||
return ServiceModel.from_row(row) if row else None
|
||
|
||
def list_all(self) -> List[ServiceModel]:
|
||
return self._select_all(self._TABLE, ServiceModel)
|
||
|
||
# ------------------------------
|
||
# Domain‑specific queries
|
||
# ------------------------------
|
||
def upcoming(self, after: Optional[date] = None, limit: int = 100) -> List[ServiceModel]:
|
||
"""
|
||
Return services that occur on or after ``after`` (defaults to today).
|
||
Results are ordered chronologically.
|
||
"""
|
||
after_date = after or date.today()
|
||
sql = f"""
|
||
SELECT *
|
||
FROM {self._TABLE}
|
||
WHERE ServiceDate >= ?
|
||
ORDER BY ServiceDate ASC
|
||
LIMIT ?
|
||
"""
|
||
rows = self.db.fetchall(sql, (after_date.isoformat(), limit))
|
||
return [ServiceModel.from_row(r) for r in rows]
|
||
|
||
def by_type(self, service_type_ids: Sequence[int]) -> List[ServiceModel]:
|
||
"""
|
||
Fetch all services whose ``ServiceTypeId`` is in the supplied list.
|
||
Empty input → empty list (no DB round‑trip).
|
||
"""
|
||
if not service_type_ids:
|
||
return []
|
||
|
||
placeholders = ",".join("?" for _ in service_type_ids)
|
||
sql = f"""
|
||
SELECT *
|
||
FROM {self._TABLE}
|
||
WHERE ServiceTypeId IN ({placeholders})
|
||
ORDER BY ServiceDate ASC
|
||
"""
|
||
rows = self.db.fetchall(sql, tuple(service_type_ids))
|
||
return [ServiceModel.from_row(r) for r in rows]
|
||
|
||
# ------------------------------
|
||
# Update helpers (optional)
|
||
# ------------------------------
|
||
def reschedule(self, service_id: int, new_date: date) -> None:
|
||
"""
|
||
Change the ``ServiceDate`` of an existing service.
|
||
"""
|
||
sql = f"""
|
||
UPDATE {self._TABLE}
|
||
SET ServiceDate = ?
|
||
WHERE {self._PK} = ?
|
||
"""
|
||
self.db.execute(sql, (new_date.isoformat(), service_id)) |