Extract API Pydantic models into schemas.py

Move the seven request/response models out of web_api.py into a dedicated
src/schemas.py (separation of concerns; first step of the file-size cleanup).
web_api.py imports them back, so behaviour is unchanged.

Note: web_api.py is still over the 500-line guideline; the remaining bulk is
the inline HTML dashboard in root(), to be extracted in a follow-up.
This commit is contained in:
2026-07-22 14:07:31 +07:00
parent 12b7f9f422
commit ad7f7b8c76
2 changed files with 79 additions and 65 deletions
+69
View File
@@ -0,0 +1,69 @@
#!/usr/bin/env python3
"""Pydantic request/response schemas for the water monitoring web API."""
from datetime import datetime
from typing import Any, Dict, Optional
from pydantic import BaseModel, Field
class StationResponse(BaseModel):
station_id: int
station_code: str
thai_name: str
english_name: str
latitude: Optional[float] = None
longitude: Optional[float] = None
geohash: Optional[str] = None
status: str = "active"
class StationCreateModel(BaseModel):
station_code: str = Field(..., description="Station code (e.g., P.1, P.20)")
thai_name: str = Field(..., description="Thai name of the station")
english_name: str = Field(..., description="English name of the station")
latitude: Optional[float] = Field(None, ge=-90, le=90, description="Latitude coordinate")
longitude: Optional[float] = Field(None, ge=-180, le=180, description="Longitude coordinate")
geohash: Optional[str] = Field(None, description="Geohash for the location")
status: str = Field("active", description="Station status")
class StationUpdateModel(BaseModel):
thai_name: Optional[str] = Field(None, description="Thai name of the station")
english_name: Optional[str] = Field(None, description="English name of the station")
latitude: Optional[float] = Field(None, ge=-90, le=90, description="Latitude coordinate")
longitude: Optional[float] = Field(None, ge=-180, le=180, description="Longitude coordinate")
geohash: Optional[str] = Field(None, description="Geohash for the location")
status: Optional[str] = Field(None, description="Station status")
class MeasurementResponse(BaseModel):
timestamp: datetime
station_code: str
station_name_en: str
station_name_th: str
water_level: float
discharge: Optional[float] = None
discharge_percent: Optional[float] = None
status: str = "active"
class HealthResponse(BaseModel):
overall_status: str
timestamp: str
checks: Dict[str, Dict[str, Any]]
class MetricsResponse(BaseModel):
counters: Dict[str, float]
gauges: Dict[str, float]
histograms: Dict[str, Dict[str, float]]
class ScrapingStatusResponse(BaseModel):
is_running: bool
last_run: Optional[datetime] = None
next_run: Optional[datetime] = None
total_runs: int = 0
successful_runs: int = 0
failed_runs: int = 0