mirror of
https://github.com/hansjone/netx.git
synced 2026-10-09 05:30:46 +08:00
oliver
d395aa7174
Add POST /v1/managed-ne/batch-hop, shared HopProxyFields form, and toolbar button next to connectivity test. Co-authored-by: Cursor <cursoragent@cursor.com>
137 lines
3.3 KiB
Python
137 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Literal
|
|
|
|
from pydantic import BaseModel, Field, field_validator
|
|
|
|
from .device_types import SUPPORTED_VENDORS
|
|
|
|
ConnectStatus = Literal["unknown", "testing", "pass", "fail"]
|
|
|
|
|
|
class ManagedNeCreate(BaseModel):
|
|
name: str = ""
|
|
vendor: str
|
|
device_type: str
|
|
ip_address: str
|
|
port: int = 22
|
|
protocol: str = "ssh"
|
|
username: str
|
|
password: str
|
|
tags: str = ""
|
|
remark: str = ""
|
|
hop_enabled: bool = False
|
|
hop_vendor: str = "zte"
|
|
hop_host: str = ""
|
|
hop_port: int = 22
|
|
hop_protocol: str = "ssh"
|
|
hop_username: str = ""
|
|
hop_password: str = ""
|
|
hop_command_template: str = ""
|
|
hop_vrf: str = ""
|
|
|
|
@field_validator("vendor")
|
|
@classmethod
|
|
def normalize_vendor(cls, v: str) -> str:
|
|
raw = str(v or "").strip()
|
|
if not raw:
|
|
raise ValueError("vendor_required")
|
|
for item in SUPPORTED_VENDORS:
|
|
if item.lower() == raw.lower():
|
|
return item
|
|
return "Other"
|
|
|
|
|
|
class ManagedNeUpdate(BaseModel):
|
|
name: str | None = None
|
|
vendor: str | None = None
|
|
device_type: str | None = None
|
|
ip_address: str | None = None
|
|
port: int | None = None
|
|
protocol: str | None = None
|
|
username: str | None = None
|
|
password: str | None = None
|
|
tags: str | None = None
|
|
remark: str | None = None
|
|
hop_enabled: bool | None = None
|
|
hop_vendor: str | None = None
|
|
hop_host: str | None = None
|
|
hop_port: int | None = None
|
|
hop_protocol: str | None = None
|
|
hop_username: str | None = None
|
|
hop_password: str | None = None
|
|
hop_command_template: str | None = None
|
|
hop_vrf: str | None = None
|
|
|
|
@field_validator("vendor")
|
|
@classmethod
|
|
def normalize_vendor_update(cls, v: str | None) -> str | None:
|
|
if v is None:
|
|
return None
|
|
raw = str(v).strip()
|
|
if not raw:
|
|
raise ValueError("vendor_required")
|
|
for item in SUPPORTED_VENDORS:
|
|
if item.lower() == raw.lower():
|
|
return item
|
|
return "Other"
|
|
|
|
|
|
class ManagedNeOut(BaseModel):
|
|
id: str
|
|
name: str
|
|
vendor: str
|
|
device_type: str
|
|
ip_address: str
|
|
port: int
|
|
protocol: str
|
|
username: str
|
|
connect_status: ConnectStatus
|
|
connect_message: str
|
|
connect_tested_at: datetime | None
|
|
tags: str
|
|
remark: str
|
|
hop_enabled: bool = False
|
|
hop_vendor: str = "zte"
|
|
hop_host: str = ""
|
|
hop_port: int = 22
|
|
hop_protocol: str = "ssh"
|
|
hop_username: str = ""
|
|
hop_command_template: str = ""
|
|
hop_vrf: str = ""
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
class ConnectTestRequest(BaseModel):
|
|
ids: list[str] = Field(min_length=1)
|
|
|
|
|
|
class HopProxyConfig(BaseModel):
|
|
"""Shared jump-host (proxy) settings applied to one or many NEs."""
|
|
|
|
hop_vendor: str = "zte"
|
|
hop_host: str
|
|
hop_port: int = 22
|
|
hop_protocol: str = "ssh"
|
|
hop_username: str
|
|
hop_password: str
|
|
hop_command_template: str = ""
|
|
hop_vrf: str = ""
|
|
|
|
|
|
class BatchHopApplyRequest(BaseModel):
|
|
ids: list[str] = Field(min_length=1)
|
|
hop: HopProxyConfig
|
|
|
|
|
|
class ImportFailure(BaseModel):
|
|
row: int
|
|
reason: str
|
|
|
|
|
|
class ImportResult(BaseModel):
|
|
inserted: int
|
|
updated: int
|
|
failed: list[ImportFailure]
|