Spaces:
Configuration error
Configuration error
| """MongoDB document model for KPI cache snapshots.""" | |
| from datetime import datetime | |
| from typing import Optional, Dict, Any | |
| from pydantic import BaseModel, Field | |
| class KPICacheDocument(BaseModel): | |
| """ | |
| Durable KPI snapshot stored in MongoDB. | |
| One document per (merchant_id, widget_id, period_window, branch_id). | |
| """ | |
| cache_id: str = Field(..., description="UUID for this cache document") | |
| merchant_id: str = Field(..., description="Tenant identifier") | |
| widget_id: str = Field(..., description="KPI widget identifier") | |
| period_window: str = Field(..., description="today | last_7_days | mtd | ytd | last_12_months") | |
| branch_id: str = Field("all", description="Branch/warehouse filter; 'all' means no filter") | |
| # KPI payload | |
| value: float = Field(..., description="Primary KPI value") | |
| unit: str = Field(..., description="count | INR | % | ratio") | |
| delta: Optional[float] = Field(None) | |
| delta_percentage: Optional[float] = Field(None) | |
| trend: str = Field("neutral", description="up | down | neutral") | |
| secondary_values: Optional[Dict[str, Any]] = Field(None) | |
| drill_down_url: Optional[str] = Field(None) | |
| # Metadata | |
| computed_at: datetime = Field(default_factory=datetime.utcnow) | |
| expires_at: Optional[datetime] = Field(None, description="When this snapshot should be rebuilt") | |
| cached: bool = Field(False, description="Always False for freshly written docs") | |
| error: Optional[str] = Field(None, description="Set when computation failed") | |