|
| 1 | +"""Device catalog endpoints. List + detail; identity-only records are unscored.""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +from typing import Annotated, Any |
| 6 | + |
| 7 | +from fastapi import APIRouter, Query |
| 8 | +from sqlalchemy import func |
| 9 | +from sqlmodel import select |
| 10 | +from sqlmodel.sql.expression import SelectOfScalar |
| 11 | + |
| 12 | +from app.dependencies import PaginationDep, SessionDep |
| 13 | +from app.errors import APIError, not_found |
| 14 | +from app.models.brand import Brand |
| 15 | +from app.models.device_catalog import DeviceCatalog |
| 16 | +from app.models.soc import SoC |
| 17 | +from app.routers.utils import build_ref_page |
| 18 | +from app.schemas.common import Page, ResourceRef |
| 19 | +from app.schemas.device_catalog import DeviceCatalogRead |
| 20 | +from app.schemas.serializers import device_catalog_read, resource_ref |
| 21 | + |
| 22 | +router = APIRouter(prefix="/device-catalog", tags=["device-catalog"]) |
| 23 | + |
| 24 | +_SORT_FIELDS: dict[str, Any] = { |
| 25 | + "name": DeviceCatalog.name, |
| 26 | + "android_sdk_min": DeviceCatalog.android_sdk_min, |
| 27 | +} |
| 28 | + |
| 29 | + |
| 30 | +def _apply_sort(stmt: SelectOfScalar[Any], sort: str | None) -> SelectOfScalar[Any]: |
| 31 | + if not sort: |
| 32 | + return stmt.order_by(DeviceCatalog.name) |
| 33 | + descending = sort.startswith("-") |
| 34 | + field = sort[1:] if descending else sort |
| 35 | + column = _SORT_FIELDS.get(field) |
| 36 | + if column is None: |
| 37 | + raise APIError(400, "INVALID_REQUEST", f"Cannot sort by '{field}'") |
| 38 | + return stmt.order_by(column.desc() if descending else column.asc()) |
| 39 | + |
| 40 | + |
| 41 | +@router.get("", summary="List device catalog entries") |
| 42 | +def list_device_catalog( |
| 43 | + session: SessionDep, |
| 44 | + pagination: PaginationDep, |
| 45 | + brand: Annotated[str | None, Query()] = None, |
| 46 | + form_factor: Annotated[str | None, Query()] = None, |
| 47 | + base_model: Annotated[str | None, Query(alias="base_model_slug")] = None, |
| 48 | + sort: Annotated[str | None, Query()] = None, |
| 49 | +) -> Page[ResourceRef]: |
| 50 | + filters = [] |
| 51 | + if brand is not None: |
| 52 | + row = session.exec(select(Brand).where(Brand.slug == brand)).first() |
| 53 | + if row is None: |
| 54 | + return build_ref_page([], count=0, path="/v1/device-catalog", pagination=pagination) |
| 55 | + filters.append(DeviceCatalog.brand_id == row.id) |
| 56 | + if form_factor is not None: |
| 57 | + filters.append(DeviceCatalog.form_factor == form_factor) |
| 58 | + if base_model is not None: |
| 59 | + filters.append(DeviceCatalog.base_model_slug == base_model) |
| 60 | + |
| 61 | + count_stmt = select(func.count()).select_from(DeviceCatalog) |
| 62 | + list_stmt = select(DeviceCatalog) |
| 63 | + for clause in filters: |
| 64 | + count_stmt = count_stmt.where(clause) |
| 65 | + list_stmt = list_stmt.where(clause) |
| 66 | + |
| 67 | + count = session.exec(count_stmt).one() |
| 68 | + list_stmt = _apply_sort(list_stmt, sort).offset(pagination.offset).limit(pagination.limit) |
| 69 | + rows = session.exec(list_stmt).all() |
| 70 | + |
| 71 | + refs = [resource_ref("device-catalog", row.slug, row.name) for row in rows] |
| 72 | + applied = { |
| 73 | + k: v |
| 74 | + for k, v in ( |
| 75 | + ("brand", brand), |
| 76 | + ("form_factor", form_factor), |
| 77 | + ("base_model_slug", base_model), |
| 78 | + ("sort", sort), |
| 79 | + ) |
| 80 | + if v |
| 81 | + } |
| 82 | + return build_ref_page( |
| 83 | + refs, count=count, path="/v1/device-catalog", pagination=pagination, filters=applied |
| 84 | + ) |
| 85 | + |
| 86 | + |
| 87 | +@router.get("/{slug}", summary="Get a device catalog entry") |
| 88 | +def get_device_catalog_entry(slug: str, session: SessionDep) -> DeviceCatalogRead: |
| 89 | + entry = session.exec(select(DeviceCatalog).where(DeviceCatalog.slug == slug)).first() |
| 90 | + if entry is None: |
| 91 | + raise not_found("Device catalog entry", slug) |
| 92 | + brand = session.get(Brand, entry.brand_id) |
| 93 | + if brand is None: # pragma: no cover - guarded by FK + validation |
| 94 | + raise not_found("Brand", str(entry.brand_id)) |
| 95 | + soc = session.get(SoC, entry.soc_id) if entry.soc_id is not None else None |
| 96 | + soc_manufacturer = session.get(Brand, soc.manufacturer_id) if soc is not None else None |
| 97 | + return device_catalog_read(entry, brand, soc, soc_manufacturer) |
0 commit comments