126 lines
3.7 KiB
Python
126 lines
3.7 KiB
Python
from datetime import datetime
|
|
from typing import Any, Optional
|
|
|
|
from pydantic import BaseModel, field_validator
|
|
|
|
from app.models.device import DEVICE_CATEGORIES, DEVICE_ROLES, DEVICE_SOURCES
|
|
|
|
|
|
class DeviceCreate(BaseModel):
|
|
ip: str
|
|
hostname: Optional[str] = None
|
|
mac: Optional[str] = None
|
|
category: str = "other"
|
|
role: str = "other"
|
|
model: Optional[str] = None
|
|
firmware_version: Optional[str] = None
|
|
source: str = "manual"
|
|
rack_id: Optional[int] = None
|
|
ssh_user_override: Optional[str] = None
|
|
ssh_password_override: Optional[str] = None # plaintext, encrypted on write
|
|
ssh_port_override: Optional[int] = None
|
|
ssh_key_id_override: Optional[int] = None
|
|
connection_params: dict[str, Any] = {}
|
|
device_meta: dict[str, Any] = {}
|
|
monitor_enabled: bool = False
|
|
notes: Optional[str] = None
|
|
is_active: bool = True
|
|
|
|
@field_validator("category")
|
|
@classmethod
|
|
def validate_category(cls, v: str) -> str:
|
|
if v not in DEVICE_CATEGORIES:
|
|
raise ValueError(f"category must be one of {DEVICE_CATEGORIES}")
|
|
return v
|
|
|
|
@field_validator("role")
|
|
@classmethod
|
|
def validate_role(cls, v: str) -> str:
|
|
if v not in DEVICE_ROLES:
|
|
raise ValueError(f"role must be one of {DEVICE_ROLES}")
|
|
return v
|
|
|
|
|
|
class DeviceUpdate(BaseModel):
|
|
ip: Optional[str] = None
|
|
hostname: Optional[str] = None
|
|
mac: Optional[str] = None
|
|
category: Optional[str] = None
|
|
role: Optional[str] = None
|
|
model: Optional[str] = None
|
|
firmware_version: Optional[str] = None
|
|
rack_id: Optional[int] = None
|
|
ssh_user_override: Optional[str] = None
|
|
ssh_password_override: Optional[str] = None
|
|
ssh_port_override: Optional[int] = None
|
|
ssh_key_id_override: Optional[int] = None
|
|
connection_params: Optional[dict[str, Any]] = None
|
|
device_meta: Optional[dict[str, Any]] = None
|
|
monitor_enabled: Optional[bool] = None
|
|
notes: Optional[str] = None
|
|
is_active: Optional[bool] = None
|
|
|
|
|
|
class DeviceRead(BaseModel):
|
|
id: int
|
|
object_id: int
|
|
ip: str
|
|
hostname: Optional[str] = None
|
|
mac: Optional[str] = None
|
|
category: str
|
|
role: str
|
|
model: Optional[str] = None
|
|
firmware_version: Optional[str] = None
|
|
source: str
|
|
external_id: Optional[str] = None
|
|
rack_id: Optional[int] = None
|
|
rack_name: Optional[str] = None
|
|
rack_type: Optional[str] = None
|
|
ssh_user_override: Optional[str] = None
|
|
ssh_port_override: Optional[int] = None
|
|
ssh_key_id_override: Optional[int] = None
|
|
ssh_last_auth: Optional[dict[str, Any]] = None
|
|
connection_params: dict[str, Any] = {}
|
|
device_meta: dict[str, Any] = {}
|
|
capabilities: list[str] = []
|
|
monitor_enabled: bool
|
|
last_ping_at: Optional[datetime] = None
|
|
last_seen: Optional[datetime] = None
|
|
last_ping_ms: Optional[int] = None
|
|
status: str
|
|
notes: Optional[str] = None
|
|
is_active: bool
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
model_config = {"from_attributes": True}
|
|
|
|
@classmethod
|
|
def from_orm_with_rack(cls, device: Any) -> "DeviceRead":
|
|
obj = cls.model_validate(device)
|
|
if device.rack is not None:
|
|
obj.rack_name = device.rack.name
|
|
obj.rack_type = device.rack.rack_type
|
|
return obj
|
|
|
|
|
|
class DeviceImportRow(BaseModel):
|
|
ip: str
|
|
hostname: Optional[str] = None
|
|
category: str = "other"
|
|
role: str = "other"
|
|
model: Optional[str] = None
|
|
mac: Optional[str] = None
|
|
notes: Optional[str] = None
|
|
|
|
|
|
class DeviceImportPreview(BaseModel):
|
|
would_create: list[DeviceImportRow] = []
|
|
would_update: list[dict] = []
|
|
errors: list[dict] = []
|
|
|
|
|
|
class DeviceImportResult(BaseModel):
|
|
created: int
|
|
updated: int
|
|
errors: list[dict] = []
|