feat(api): type link filters and available_filters response

This commit is contained in:
Mark Paronyan committed 2026-10-07 15:48:52 +03:00
1 parent 2fb7056cb9
commit 11a310dd9b
5 files changed
+186 -6

No files matched your search

+7 -2
View File
@@ -123,8 +123,13 @@ async def create_link(project_id: UUID, link_create: schemas.LinkCreate) -> sche
return link.asdict()
@router.get("/{link_id}/available_filters", dependencies=[Depends(has_privilege("Link.Audit"))])
async def get_filters(link: Link = Depends(dep_link)) -> List[dict]:
@router.get(
"/{link_id}/available_filters",
response_model=List[schemas.LinkFilterDefinition],
response_model_exclude_unset=True,
dependencies=[Depends(has_privilege("Link.Audit"))],
)
async def get_filters(link: Link = Depends(dep_link)) -> List[schemas.LinkFilterDefinition]:
"""
Return all filters available for a given link.
+8
View File
@@ -37,6 +37,10 @@ from .controller.links import (
Link,
LinkCapture,
LinkCreate,
LinkFilterDefinition,
LinkFilterParameter,
LinkFilters,
LinkFilterType,
LinkUpdate,
MarkerCreate,
MarkerDefinitionCreate,
@@ -244,6 +248,10 @@ __all__ = [
"Link",
"LinkCapture",
"LinkCreate",
"LinkFilterDefinition",
"LinkFilterParameter",
"LinkFilterType",
"LinkFilters",
"LinkUpdate",
"LoggedInUserUpdate",
"MarkerCreate",
+74 -3
View File
@@ -15,10 +15,10 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from enum import Enum
from typing import List, Optional, Tuple
from typing import Annotated, List, Literal, Optional, Tuple
from uuid import UUID, uuid4
from pydantic import BaseModel, Field, field_validator
from pydantic import BaseModel, ConfigDict, Field, field_validator
from ..update import PartialUpdateModel
from .labels import Label
@@ -54,6 +54,77 @@ class LinkStyle(BaseModel):
control_offset: Optional[Tuple[float, float]] = None
class LinkFilterType(str, Enum):
"""
Packet filter type.
"""
frequency_drop = "frequency_drop"
packet_loss = "packet_loss"
delay = "delay"
corrupt = "corrupt"
bpf = "bpf"
class LinkFilters(BaseModel):
"""
Packet filters applied on a link. Each filter is an array of positional values.
A filter set to an empty array or to its disabled value (0, 0 for delay, empty text for bpf) is not applied.
"""
model_config = ConfigDict(extra="forbid")
frequency_drop: Optional[List[Annotated[int, Field(ge=-1, le=32767)]]] = Field(
None,
max_length=1,
description="[frequency]: -1 drops every packet, N > 0 drops every Nth packet, 0 disables the filter",
)
packet_loss: Optional[List[Annotated[int, Field(ge=0, le=100)]]] = Field(
None,
max_length=1,
description="[chance]: percentage chance for a packet to be lost, 0 disables the filter",
)
delay: Optional[List[Annotated[int, Field(ge=0, le=32767)]]] = Field(
None,
max_length=2,
description="[latency, jitter] in milliseconds: latency must be 1 to 32767 unless both values are 0, "
"which disables the filter; jitter is 0 to 32767",
)
corrupt: Optional[List[Annotated[int, Field(ge=0, le=100)]]] = Field(
None,
max_length=1,
description="[chance]: percentage chance for a packet to be corrupted, 0 disables the filter",
)
bpf: Optional[List[str]] = Field(
None,
max_length=1,
description="[expressions]: BPF expressions, one per line, matching packets are dropped",
)
class LinkFilterParameter(BaseModel):
"""
Parameter of a packet filter.
"""
name: str
type: Literal["int", "text"]
minimum: Optional[int] = None
maximum: Optional[int] = None
unit: Optional[str] = None
class LinkFilterDefinition(BaseModel):
"""
Packet filter available on a link.
"""
type: LinkFilterType
name: str
description: str
parameters: List[LinkFilterParameter]
class LinkBase(BaseModel):
"""
Link data.
@@ -62,7 +133,7 @@ class LinkBase(BaseModel):
nodes: Optional[List[LinkNode]] = Field(None, min_length=0, max_length=2)
suspend: Optional[bool] = None
link_style: Optional[LinkStyle] = None
filters: Optional[dict] = None
filters: Optional[LinkFilters] = None
markers: Optional[dict] = Field(
None, description="Traffic-insight markers on this link: name → {bpf, tag, enabled}"
)
+6 -1
View File
@@ -32,6 +32,11 @@ from .nodes import Node
from .projects import Supplier, Variable
class TopologyLink(Link):
# Legacy topology files store filters as untyped dicts
filters: Optional[dict] = None # type: ignore[assignment]
class TopologyType(str, Enum):
topology = "topology"
@@ -39,7 +44,7 @@ class TopologyType(str, Enum):
class TopologyContent(BaseModel):
computes: List[Compute] = Field(..., description="List of computes")
drawings: List[Drawing] = Field(..., description="List of drawings")
links: List[Link] = Field(..., description="List of links")
links: List[TopologyLink] = Field(..., description="List of links")
nodes: List[Node] = Field(..., description="List of nodes")