Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fixing HAInfo type #12

Merged
merged 2 commits into from
Sep 20, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 0 additions & 13 deletions pa_api/xmlapi/types/config/interface.py
Original file line number Diff line number Diff line change
Expand Up @@ -195,13 +195,6 @@ def flatten(self, parent=None) -> Iterable[GenericInterface]:
if self.layer3:
yield from self.layer3.flatten(parent)

# @model_validator(mode="before")
# @classmethod
# def test(cls, data):
# print("\n" * 5)
# print("Ethernet:", data)
# return data


class AggregateEthernet(XMLBaseModel):
model_config = ConfigDict(extra="allow")
Expand Down Expand Up @@ -229,12 +222,6 @@ class AggregateEthernet(XMLBaseModel):
validation_alias=AliasPath("tag", "member"),
default_factory=list,
)
# @model_validator(mode="before")
# @classmethod
# def test(cls, data):
# print("\n" * 5)
# print("AggregateEthernet:", data)
# return data

def flatten(self, parent=None) -> Iterable[GenericInterface]:
if self.name:
Expand Down
3 changes: 2 additions & 1 deletion pa_api/xmlapi/types/operations/__init__.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from . import device, job, software
from .device import Device, HAInfo, VPNFlow
from .device import Device, VPNFlow
from .ha import HAGroup, HAInfo, HALocalInfo, HAPeerInfo
from .job import Job, JobResult
from .software import SoftwareVersion
from .system import SystemInfo
62 changes: 1 addition & 61 deletions pa_api/xmlapi/types/operations/device.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,14 +2,7 @@

from pydantic import AliasPath, ConfigDict, Field

from pa_api.utils import (
first,
)
from pa_api.xmlapi.types.utils import (
Datetime,
XMLBaseModel,
mksx,
)
from pa_api.xmlapi.types.utils import Datetime, XMLBaseModel


class Device(XMLBaseModel):
Expand Down Expand Up @@ -119,56 +112,3 @@ class VPNFlow(XMLBaseModel):
state: str
mon: str
owner: str


class HAInfo(XMLBaseModel):
model_config = ConfigDict(populate_by_name=True, extra="allow")

enabled: bool
preemptive: Optional[bool] = None
mode: Optional[str] = None
state: Optional[str] = None
peer_state: Optional[str] = None
priority: Optional[int] = None
peer_priority: Optional[int] = None
is_primary: Optional[bool] = None
peer_conn_status: Optional[str] = None
mgmt_ip: Optional[str] = None
ha1_ipaddr: Optional[str] = None
ha1_backup_ipaddr: Optional[str] = None
ha2_ipaddr: Optional[str] = None
ha1_macaddress: Optional[str] = None
ha1_backup_macaddress: Optional[str] = None
ha2_macaddress: Optional[str] = None

@classmethod
def from_xml(cls, xml):
# TODO: Use correct pydantic functionalities
if isinstance(xml, (list, tuple)):
xml = first(xml)
if xml is None:
return None
p = mksx(xml)
priority = p("./group/local-info/priority/text()", parser=int)
peer_priority = p("./group/peer-info/priority/text()", parser=int)
is_primary = None
if priority is not None and peer_priority is not None:
is_primary = priority < peer_priority
return HAInfo(
enabled=p("./enabled/text()"),
preemptive=p("./group/local-info/preemptive/text()"),
mode=p("./group/local-info/mode/text()"),
state=p("./group/local-info/state/text()"),
peer_state=p("./group/peer-info/state/text()"),
priority=priority,
peer_priority=peer_priority,
is_primary=is_primary,
peer_conn_status=p("./group/peer-info/conn-status/text()"),
mgmt_ip=p("./group/local-info/mgmt-ip/text()"),
ha1_ipaddr=p("./group/local-info/ha1-ipaddr/text()"),
ha1_backup_ipaddr=p("./group/local-info/ha1-backup-ipaddr/text()"),
ha2_ipaddr=p("./group/local-info/ha2-ipaddr/text()"),
ha1_macaddress=p("./group/local-info/ha1-macaddr/text()"),
ha1_backup_macaddress=p("./group/local-info/ha1-backup-macaddr/text()"),
ha2_macaddress=p("./group/local-info/ha2-macaddr/text()"),
)
47 changes: 47 additions & 0 deletions pa_api/xmlapi/types/operations/ha.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
from typing import Optional

from pydantic import ConfigDict, Field

from pa_api.xmlapi.types.utils import String, XMLBaseModel


class HALocalInfo(XMLBaseModel):
model_config = ConfigDict(populate_by_name=True, extra="allow")

priority: int
preemptive: bool
state: str
mode: String
ha1_backup_macaddr: str = Field(alias="ha1-backup-macaddr")
ha1_macaddr: str = Field(alias="ha1-macaddr")
mgmt_ip: str = Field(alias="mgmt-ip")


class HAPeerInfo(XMLBaseModel):
model_config = ConfigDict(populate_by_name=True, extra="allow")
priority: int
preemptive: bool
state: str
mode: str
conn_status: str = Field(alias="conn-status")
ha1_backup_macaddr: str = Field(alias="ha1-backup-macaddr")
ha1_macaddr: str = Field(alias="ha1-macaddr")
mgmt_ip: str = Field(alias="mgmt-ip")


class HAGroup(XMLBaseModel):
model_config = ConfigDict(populate_by_name=True, extra="allow")

local_info: HALocalInfo = Field(alias="local-info")
peer_info: HAPeerInfo = Field(alias="peer-info")

@property
def is_primary(self):
return self.local_info.priority < self.peer_info.priority


class HAInfo(XMLBaseModel):
model_config = ConfigDict(populate_by_name=True, extra="allow")

enabled: bool
group: Optional[HAGroup]
10 changes: 2 additions & 8 deletions pa_api/xmlapi/types/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,8 @@
class XMLBaseModel(BaseModel):
@classmethod
def from_xml(cls, xml) -> Self:
rulebase = first(el2dict(xml).values())
return cls.model_validate(rulebase)
data = first(el2dict(xml).values())
return cls.model_validate(data)


def parse_datetime(d):
Expand Down Expand Up @@ -95,12 +95,6 @@ def ensure_list(v: Any) -> typing.List[Any]:
# Similar to typing.List, but ensure to always return a list
List = Annotated[typing.List[Element], BeforeValidator(ensure_list)]

# from pydantic import TypeAdapter
# ta = TypeAdapter(List[int])
# print(ta.validate_python(5))
# print(ta.validate_python([5]))
# print(ta.validate_python())


def xml_text(v: Any):
if isinstance(v, dict) and "#text" in v:
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[tool.poetry]
name = "pa-api-sdk"
version = "0.1.3"
version = "0.1.4"
description = ""
authors = ["David Gallay <[email protected]>"]
readme = "README.md"
Expand Down
Loading