from datetime import datetime from typing import Any, Dict, Generic, List, Literal, Optional, TypeVar from pydantic import BaseModel, Field T = TypeVar('T') JobStatus = Literal['pending', 'running', 'completed', 'failed', 'cancelled'] ObjectStatus = Literal['active', 'planned', 'inactive', 'retired', 'unknown'] ObjectType = Literal[ 'site', 'area', 'rack', 'physical_host', 'vm', 'lxc', 'container', 'app', 'switch', 'firewall', 'router', 'ap', 'camera', 'storage', 'other' ] RelationType = Literal[ 'contains', 'mounted_in', 'attached_to', 'connects_to', 'uplink_to', 'routes_via', 'bridges_to', 'depends_on', 'hosts_service', 'managed_by' ] RelationLayer = Literal['physical', 'network', 'logical', 'service'] RelationSource = Literal['manual', 'snmp', 'lldp', 'arp', 'nmap', 'inference', 'import'] RelationStatus = Literal['active', 'stale', 'planned', 'deleted'] RelationDirection = Literal['forward'] class HealthCheck(BaseModel): status: Literal['pass', 'warn', 'fail'] latency_ms: Optional[float] = None detail: Optional[str] = None class HealthResponse(BaseModel): status: Literal['pass', 'warn', 'fail'] service: str version: Optional[str] = None time: datetime uptime_s: Optional[float] = None checks: Dict[str, HealthCheck] = Field(default_factory=dict) # --------------------------------------------------------------------------- # NetObject # --------------------------------------------------------------------------- class NetObject(BaseModel): id: str name: str slug: Optional[str] = None type: ObjectType parent_id: Optional[str] = None status: ObjectStatus vendor: Optional[str] = None model: Optional[str] = None serial: Optional[str] = None notes: Optional[str] = None created_at: datetime updated_at: datetime class NetObjectCreate(BaseModel): name: str type: ObjectType parent_id: Optional[str] = None status: ObjectStatus = 'active' vendor: Optional[str] = None model: Optional[str] = None serial: Optional[str] = None notes: Optional[str] = None class NetObjectUpdate(BaseModel): name: Optional[str] = None type: Optional[ObjectType] = None parent_id: Optional[str] = None status: Optional[ObjectStatus] = None vendor: Optional[str] = None model: Optional[str] = None serial: Optional[str] = None notes: Optional[str] = None # --------------------------------------------------------------------------- # IPAddress / Subnet / Link (read-only for now) # --------------------------------------------------------------------------- class IPAddress(BaseModel): id: str address: str version: Literal[4, 6] subnet_id: Optional[str] = None object_id: Optional[str] = None interface_id: Optional[str] = None hostname: Optional[str] = None source: Literal['manual', 'snmp', 'arp', 'lldp', 'nmap', 'import', 'other'] status: Literal['assigned', 'reserved', 'free', 'unknown'] = 'unknown' class Subnet(BaseModel): id: str name: str cidr: str gateway: Optional[str] = None vlan_id: Optional[int] = None notes: Optional[str] = None class Link(BaseModel): id: str source_object_id: str target_object_id: str source_interface_id: Optional[str] = None target_interface_id: Optional[str] = None type: Literal['physical', 'logical', 'hosted_on', 'member_of', 'uplink', 'other'] confidence: float = 1.0 source: Literal['manual', 'snmp', 'lldp', 'arp', 'import', 'inference'] # --------------------------------------------------------------------------- # Relation # --------------------------------------------------------------------------- class Relation(BaseModel): id: str type: RelationType layer: RelationLayer source_id: str target_id: str direction: RelationDirection = 'forward' source: RelationSource confidence: float = Field(ge=0, le=1, default=1.0) status: RelationStatus = 'active' attributes: Dict[str, Any] = Field(default_factory=dict) created_at: datetime updated_at: datetime class RelationCreate(BaseModel): type: RelationType layer: RelationLayer source_id: str target_id: str direction: RelationDirection = 'forward' source: RelationSource = 'manual' confidence: float = Field(ge=0, le=1, default=1.0) status: RelationStatus = 'active' attributes: Dict[str, Any] = Field(default_factory=dict) class RelationUpdate(BaseModel): type: Optional[RelationType] = None layer: Optional[RelationLayer] = None source_id: Optional[str] = None target_id: Optional[str] = None direction: Optional[RelationDirection] = None source: Optional[RelationSource] = None confidence: Optional[float] = Field(ge=0, le=1, default=None) status: Optional[RelationStatus] = None attributes: Optional[Dict[str, Any]] = None # --------------------------------------------------------------------------- # Tree / Context # --------------------------------------------------------------------------- class InventoryTreeNode(BaseModel): id: str name: str type: ObjectType status: ObjectStatus badges: List[str] = Field(default_factory=list) children_count: int = 0 children: List['InventoryTreeNode'] = Field(default_factory=list) class NodeContextRelations(BaseModel): incoming: List[Relation] = Field(default_factory=list) outgoing: List[Relation] = Field(default_factory=list) class NodeContext(BaseModel): item: NetObject parent: Optional[NetObject] = None children: List[NetObject] = Field(default_factory=list) relations: NodeContextRelations = Field(default_factory=NodeContextRelations) # --------------------------------------------------------------------------- # Jobs / Discovery # --------------------------------------------------------------------------- class Job(BaseModel): id: str type: str status: JobStatus progress: int = 0 message: Optional[str] = None error: Optional[str] = None payload: Optional[Dict[str, Any]] = None result: Optional[Dict[str, Any]] = None created_at: datetime started_at: Optional[datetime] = None finished_at: Optional[datetime] = None class DiscoveryProfile(BaseModel): id: str name: str strategy: Literal['fast', 'balanced', 'deep', 'custom'] use_icmp: bool = True use_arp: bool = True use_snmp: bool = False use_lldp: bool = False use_portscan: bool = False notes: Optional[str] = None class DiscoveryRun(BaseModel): id: str profile_id: str subnet_ids: List[str] = Field(default_factory=list) job_id: Optional[str] = None status: JobStatus summary: Optional[Dict[str, Any]] = None created_at: datetime started_at: Optional[datetime] = None finished_at: Optional[datetime] = None # --------------------------------------------------------------------------- # Generic wrappers # --------------------------------------------------------------------------- class ListResponse(BaseModel, Generic[T]): count: int items: List[T] page: Optional[int] = None page_size: Optional[int] = None total: Optional[int] = None class EntityResponse(BaseModel, Generic[T]): item: T InventoryTreeNode.model_rebuild()