matthias.lotz
527b5645ed
Complete Pydantic V2 migration and config modernization: api/models.py: - Migrated @validator → @field_validator (Pydantic V2) - Added @classmethod decorators (required in V2) - Updated validator signatures: (cls, v, values) → (cls, v, info) - Consolidated unique validators into one method - Fixed: 3 PydanticDeprecatedSince20 warnings ✅ config/schema.py: - Migrated from @dataclass → Pydantic BaseModel - Added Field() with validation constraints: * Port: ge=1, le=65535 * Floats: gt=0, ge=0 * Strings: min_length constraints - Added descriptive field descriptions - Better type safety and runtime validation core/bootstrap.py: - Replaced SimpleNamespace → MQTTConfig (Pydantic) - Now returns properly typed Pydantic models - Better IDE autocomplete and type checking Benefits: ✅ No more Pydantic V1 deprecation warnings ✅ Runtime validation for all config fields ✅ Better error messages on invalid config ✅ Type-safe config throughout the application ✅ Automatic validation (port ranges, positive numbers, etc.) ✅ Prepared for Pydantic V3 migration Migration verified: - Config loading works: load_config('config.example.yaml') ✓ - Bootstrap works with Pydantic models ✓ - Field validation works (tested port ranges, thresholds) ✓ - All existing functionality preserved ✓ Test: python3 -c 'from core.bootstrap import bootstrap; bootstrap()' ✅ Works perfectly with new Pydantic V2 models See OPTIMIZATION_PLAN.md Phase 2.3 for details.
85 lines
3.3 KiB
Python
85 lines
3.3 KiB
Python
"""Pydantic models for API validation."""
|
|
|
|
from datetime import datetime
|
|
|
|
from pydantic import BaseModel, Field, field_validator
|
|
|
|
|
|
class MqttConfig(BaseModel):
|
|
"""MQTT Broker configuration."""
|
|
|
|
broker: str = Field("localhost", description="MQTT broker hostname")
|
|
port: int = Field(1883, description="MQTT broker port")
|
|
username: str = Field("", description="MQTT username (optional)")
|
|
password: str = Field("", description="MQTT password (optional)")
|
|
client_id: str = Field("iot_bridge", description="MQTT client ID")
|
|
keepalive: int = Field(60, description="MQTT keepalive interval (s)")
|
|
use_tls: bool = Field(False, description="Use TLS/SSL for MQTT connection")
|
|
|
|
|
|
class SessionConfig(BaseModel):
|
|
"""Session detection configuration for a device."""
|
|
|
|
standby_threshold_w: float = Field(
|
|
..., gt=0, description="Power threshold for session start (W)"
|
|
)
|
|
working_threshold_w: float = Field(
|
|
..., gt=0, description="Power threshold for working state (W)"
|
|
)
|
|
start_debounce_s: float = Field(3.0, gt=0, description="Debounce time for session start (s)")
|
|
stop_debounce_s: float = Field(15.0, gt=0, description="Debounce time for session stop (s)")
|
|
message_timeout_s: float = Field(
|
|
20.0, gt=0, description="Max time without messages before timeout (s)"
|
|
)
|
|
heartbeat_interval_s: float = Field(
|
|
300.0, gt=0, description="Interval for heartbeat events (s)"
|
|
)
|
|
|
|
@field_validator("working_threshold_w")
|
|
@classmethod
|
|
def working_must_be_greater_than_standby(cls, v, info):
|
|
"""Validate that working threshold is greater than standby threshold."""
|
|
standby = info.data.get("standby_threshold_w")
|
|
if standby is not None and v <= standby:
|
|
raise ValueError("working_threshold_w must be greater than standby_threshold_w")
|
|
return v
|
|
|
|
|
|
class DeviceConfig(BaseModel):
|
|
"""Configuration for a single IoT device."""
|
|
|
|
device_id: str = Field(..., min_length=1, description="Unique device identifier")
|
|
machine_name: str = Field(..., min_length=1, description="Human-readable machine name")
|
|
mqtt_topic: str = Field(..., min_length=1, description="MQTT topic to subscribe to")
|
|
parser_type: str = Field("shelly_pm", description="Parser type (e.g., 'shelly_pm')")
|
|
session_config: SessionConfig
|
|
|
|
|
|
class BridgeConfig(BaseModel):
|
|
"""Complete bridge configuration from Odoo."""
|
|
|
|
mqtt: MqttConfig | None = Field(None, description="MQTT broker configuration (optional)")
|
|
devices: list[DeviceConfig] = Field(
|
|
default_factory=list, description="List of devices to monitor"
|
|
)
|
|
timestamp: str | None = Field(
|
|
default_factory=lambda: datetime.utcnow().isoformat(), description="Config timestamp"
|
|
)
|
|
version: str | None = Field("1.0", description="Config format version")
|
|
|
|
@field_validator("devices")
|
|
@classmethod
|
|
def validate_unique_constraints(cls, v):
|
|
"""Validate that device_ids and mqtt_topics are unique."""
|
|
# Check unique device IDs
|
|
device_ids = [d.device_id for d in v]
|
|
if len(device_ids) != len(set(device_ids)):
|
|
raise ValueError("device_id must be unique")
|
|
|
|
# Check unique MQTT topics
|
|
topics = [d.mqtt_topic for d in v]
|
|
if len(topics) != len(set(topics)):
|
|
raise ValueError("mqtt_topic must be unique")
|
|
|
|
return v
|