"""Pydantic request/response models for the FastAPI layer.""" from __future__ import annotations from typing import List, Optional from pydantic import BaseModel, Field # --------------------------------------------------------------------------- # Shared # --------------------------------------------------------------------------- class ProductOut(BaseModel): image_id: str image_url: Optional[str] = None image_urls: List[str] = Field(default_factory=list) brand: str product_name: str title: Optional[str] = None category: Optional[str] = None description: Optional[str] = None price_range: Optional[str] = None size_variants: List[str] = Field(default_factory=list) providers: List[str] = Field(default_factory=list) highlights: List[str] = Field(default_factory=list) nutrients: List[str] = Field(default_factory=list) fssai_license: Optional[str] = None product_sku: Optional[str] = None sku_source: Optional[str] = None hsn_code: Optional[str] = None final_selling_price: Optional[float] = None selling_price: Optional[float] = None barcode: Optional[str] = None barcode_type: Optional[str] = None class SourceProductOut(BaseModel): image_id: str image_url: Optional[str] = None image_urls: List[str] = Field(default_factory=list) brand: str product_name: str title: Optional[str] = None category: Optional[str] = None description: Optional[str] = None price_range: Optional[str] = None size_variants: List[str] = Field(default_factory=list) providers: List[str] = Field(default_factory=list) highlights: List[str] = Field(default_factory=list) nutrients: List[str] = Field(default_factory=list) fssai_license: Optional[str] = None product_sku: Optional[str] = None sku_source: Optional[str] = None hsn_code: Optional[str] = None final_selling_price: Optional[float] = None selling_price: Optional[float] = None barcode: Optional[str] = None barcode_type: Optional[str] = None similarity: float # --------------------------------------------------------------------------- # Health # --------------------------------------------------------------------------- class AuthConfigOut(BaseModel): """ The effective auth configuration, reported by /api/health. Unauthenticated on purpose. The failure this exists to diagnose is "nobody can sign in", so anything gated behind an admin token is unreachable exactly when it is needed. Nothing here is a secret: the admin username is already the documented one, allow_any_login=true is a fact an operator urgently needs (and an attacker discovers with a single login attempt anyway), and the fingerprint is a truncated hash of a salted digest, not a password. What it buys is a one-command answer to "is this deployment running the config I think it is?" - compare the fingerprint here against the one printed by scripts/make_auth_secrets.py --fingerprint. """ enabled: bool allow_any_login: bool admin_username: str password_hash_valid: bool password_hash_iterations: Optional[int] = None password_hash_fingerprint: str # "process-env" | "env-file" | "default" - which one actually won. admin_username_source: str password_hash_source: str class HealthOut(BaseModel): status: str database: bool ollama: bool ollama_model: str embeddings_model: str auth: AuthConfigOut # --------------------------------------------------------------------------- # Brands / catalog browsing # --------------------------------------------------------------------------- class BrandsOut(BaseModel): brands: List[str] class BrandCardOut(BaseModel): """A brand as shown on the home page card grid. `name` is the same string GET /api/brands returns, so the frontend can pass it straight back to /api/brands/{brand}/products. """ name: str slug: str product_count: int category_count: int categories: List[str] = Field(default_factory=list) image_url: Optional[str] = None initials: str class BrandCardsOut(BaseModel): total_brands: int total_products: int brands: List[BrandCardOut] class CategoriesOut(BaseModel): brand: str categories: List[str] class ProductListOut(BaseModel): brand: str total: int limit: int offset: int products: List[ProductOut] class AllProductsOut(BaseModel): total: int limit: int offset: int products: List[ProductOut] # --------------------------------------------------------------------------- # Semantic search (retrieval only, no LLM generation) # --------------------------------------------------------------------------- class SearchOut(BaseModel): query: str # Echoes the *request* param, as it always has. The brand inferred from the # query text goes in `detected_brand` instead - repurposing this field would # break any consumer reading it as "the filter I sent". brand: Optional[str] = None results: List[SourceProductOut] # Exact in brand_catalog mode. None in hybrid mode: the merge happens in # Python across N brand tables after per-table LIMITs, so there is no cheap # exact count and inventing one would misreport how much was found. total: Optional[int] = None limit: int = 0 offset: int = 0 match_mode: str = "hybrid" # "brand_catalog" | "hybrid" detected_brand: Optional[str] = None detected_category: Optional[str] = None class SuggestionOut(BaseModel): """One row in the search box's autocomplete dropdown.""" type: str # "brand" | "category" value: str # what the search box / filter should use label: str # display text sublabel: Optional[str] = None # e.g. "128 products" score: float = 0.0 class SuggestOut(BaseModel): query: str suggestions: List[SuggestionOut] # --------------------------------------------------------------------------- # RAG chat # --------------------------------------------------------------------------- class ChatTurn(BaseModel): role: str = Field(..., description="'user' or 'assistant'") content: str class ChatRequest(BaseModel): query: str = Field(..., min_length=1, max_length=2000) brand: Optional[str] = Field(None, description="Restrict retrieval to a single brand") category: Optional[str] = Field(None, description="Restrict retrieval to a category") top_k: Optional[int] = Field(None, ge=1, le=15) history: Optional[List[ChatTurn]] = Field(default=None, description="Prior turns for follow-up questions") class ChatResponseOut(BaseModel): answer: str query: str brand: Optional[str] = None detected_category: Optional[str] = Field( None, description="Product category auto-detected from the query and used to scope retrieval, if any" ) sources: List[SourceProductOut] # --------------------------------------------------------------------------- # Catalog generation (admin/ingestion trigger) # --------------------------------------------------------------------------- class CatalogGenerateRequest(BaseModel): brand: str = Field(..., min_length=1, max_length=120) max_products: int = Field(50, ge=1, le=300) class CatalogJobOut(BaseModel): job_id: str brand: str status: str detail: Optional[str] = None