netx/netx_api/topology_schemas.py
oliver 8cf1696d8c Improve topology discovery UX with review modals and live search.
Report unmatched/stub neighbors and matched links, move result lists into detail dialogs, deep-link edges to port traffic, and add fuzzy type-ahead canvas/palette NE search.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-01 13:35:52 +08:00

145 lines
3.7 KiB
Python

"""Pydantic schemas for topology maps / nodes / edges."""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, Field
class TopologyMapCreate(BaseModel):
name: str = Field(min_length=1, max_length=256)
remark: str = Field(default="", max_length=1024)
class TopologyMapUpdate(BaseModel):
name: str | None = Field(default=None, min_length=1, max_length=256)
remark: str | None = Field(default=None, max_length=1024)
class TopologyMapOut(BaseModel):
id: str
name: str
remark: str
node_count: int = 0
edge_count: int = 0
created_at: datetime | None = None
updated_at: datetime | None = None
class TopologyNodeIn(BaseModel):
id: str = Field(min_length=1, max_length=64)
managed_ne_id: str = ""
ume_ne_id: str = ""
label: str = ""
x: float = 0.0
y: float = 0.0
created_at: datetime | None = None
class TopologyEdgeIn(BaseModel):
id: str = Field(min_length=1, max_length=64)
source_node_id: str = Field(min_length=1, max_length=64)
target_node_id: str = Field(min_length=1, max_length=64)
source_port: str = ""
target_port: str = ""
source: str = "manual"
stroke_color: str = Field(default="", max_length=32)
stroke_width: int = Field(default=0, ge=0, le=12)
line_style: str = Field(default="", max_length=16)
discovered_at: datetime | None = None
created_at: datetime | None = None
class TopologyNodeOut(BaseModel):
id: str
map_id: str
managed_ne_id: str = ""
ume_ne_id: str = ""
label: str = ""
x: float = 0.0
y: float = 0.0
ne_name: str = ""
ne_ip: str = ""
vendor: str = ""
protocol: str = ""
connect_status: str = ""
class TopologyEdgeOut(BaseModel):
id: str
map_id: str
source_node_id: str
target_node_id: str
source_port: str = ""
target_port: str = ""
source: str = "manual"
stroke_color: str = ""
stroke_width: int = 0
line_style: str = ""
discovered_at: datetime | None = None
class TopologyGraphOut(BaseModel):
map: TopologyMapOut
nodes: list[TopologyNodeOut]
edges: list[TopologyEdgeOut]
class TopologyGraphPut(BaseModel):
nodes: list[TopologyNodeIn] = Field(default_factory=list)
edges: list[TopologyEdgeIn] = Field(default_factory=list)
class TopologyDiscoverRequest(BaseModel):
"""Run LLDP/CDP discovery for managed NEs currently on the map."""
protocol: str = Field(default="auto", description="auto | lldp | cdp")
ne_ids: list[str] | None = None
class TopologyDiscoverUnmatched(BaseModel):
remote_name: str = ""
remote_ip: str = ""
local_port: str = ""
remote_port: str = ""
class TopologyDiscoverLink(BaseModel):
peer_node_id: str = ""
peer_ne_id: str = ""
peer_name: str = ""
peer_ip: str = ""
local_port: str = ""
remote_port: str = ""
protocol: str = ""
action: str = "" # added | updated | kept_manual
class TopologyDiscoverNeResult(BaseModel):
ne_id: str
ne_name: str = ""
ne_ip: str = ""
ok: bool = False
command: str = ""
neighbors: int = 0
edges_added: int = 0
edges_updated: int = 0
unmatched_count: int = 0
unmatched: list[TopologyDiscoverUnmatched] = Field(default_factory=list)
links: list[TopologyDiscoverLink] = Field(default_factory=list)
parser_key: str = ""
parser_stub: bool = False
error: str = ""
raw_preview: str = ""
class TopologyDiscoverOut(BaseModel):
map_id: str
protocol: str
scanned: int = 0
edges_added: int = 0
edges_updated: int = 0
edges_stale: int = 0
results: list[TopologyDiscoverNeResult] = Field(default_factory=list)
graph: TopologyGraphOut | None = None