5c6a2e78cf
Register all active platform services and base features in core DB so admin can manage them, add delivery/hospitality/sports-center backend phases, update apps catalog and production deploy tooling. Co-authored-by: Cursor <cursoragent@cursor.com>
63 lines
2.1 KiB
Python
63 lines
2.1 KiB
Python
"""Driver query specifications — Phase 10.1."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import Select, asc, desc, or_
|
|
from sqlalchemy.sql import ColumnElement
|
|
|
|
from app.models.drivers import Driver
|
|
from app.models.types import DriverStatus
|
|
|
|
ALLOWED_SORT_FIELDS = frozenset(
|
|
{"created_at", "updated_at", "code", "display_name", "status", "mobile"}
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class DriverListSpec:
|
|
status: DriverStatus | None = None
|
|
organization_id: UUID | None = None
|
|
hub_id: UUID | None = None
|
|
q: str | None = None
|
|
sort_by: str = "created_at"
|
|
sort_dir: str = "desc"
|
|
|
|
def filter_clauses(self) -> list[ColumnElement]:
|
|
clauses: list[ColumnElement] = []
|
|
if self.status is not None:
|
|
clauses.append(Driver.status == self.status)
|
|
if self.organization_id is not None:
|
|
clauses.append(Driver.organization_id == self.organization_id)
|
|
if self.hub_id is not None:
|
|
clauses.append(Driver.hub_id == self.hub_id)
|
|
if self.q:
|
|
term = f"%{self.q.strip()}%"
|
|
clauses.append(
|
|
or_(
|
|
Driver.code.ilike(term),
|
|
Driver.display_name.ilike(term),
|
|
Driver.mobile.ilike(term),
|
|
Driver.email.ilike(term),
|
|
Driver.first_name.ilike(term),
|
|
Driver.last_name.ilike(term),
|
|
)
|
|
)
|
|
return clauses
|
|
|
|
def apply(self, stmt: Select) -> Select:
|
|
clauses = self.filter_clauses()
|
|
if clauses:
|
|
stmt = stmt.where(*clauses)
|
|
sort_key = self.sort_by if self.sort_by in ALLOWED_SORT_FIELDS else "created_at"
|
|
column = getattr(Driver, sort_key)
|
|
order = desc(column) if self.sort_dir.lower() != "asc" else asc(column)
|
|
return stmt.order_by(order)
|
|
|
|
def apply_filters_only(self, stmt: Select) -> Select:
|
|
clauses = self.filter_clauses()
|
|
if clauses:
|
|
stmt = stmt.where(*clauses)
|
|
return stmt
|