Remove casting validators

CiscoDevNet · Feb 26, 2024 · 3ad0e51 · 3ad0e51
2 parents 20d74a6 + cc47fef
commit 3ad0e51
Show file tree

Hide file tree

Showing 20 changed files with 255 additions and 287 deletions.
diff --git a/catalystwan/api/template_api.py b/catalystwan/api/template_api.py
@@ -382,7 +382,9 @@ def _edit_feature_template(self, template: FeatureTemplate, data: FeatureTemplat
         if self.is_created_by_generator(template):
             debug = False
             schema = self.get_feature_template_schema(template, debug)
-            payload = self.generate_feature_template_payload(template, schema, debug).model_dump(by_alias=True)
+            payload = self.generate_feature_template_payload(template, schema, debug).model_dump(
+                by_alias=True, mode="json"
+            )
         else:
             payload = json.loads(template.generate_payload(self.session))
 
@@ -541,7 +543,7 @@ def create_by_generator(self, template: FeatureTemplate, debug: bool) -> str:
         payload = self.generate_feature_template_payload(template, schema, debug)
 
         endpoint = "/dataservice/template/feature"
-        response = self.session.post(endpoint, json=payload.model_dump(by_alias=True, exclude_none=True))
+        response = self.session.post(endpoint, json=payload.model_dump(by_alias=True, exclude_none=True, mode="json"))
 
         return response.json()["templateId"]
 
@@ -557,9 +559,9 @@ def generate_feature_template_payload(
         )  # type: ignore
 
         fr_template_fields = [FeatureTemplateField(**field) for field in schema["fields"]]  # TODO
-
+        json_dumped_template = template.model_dump(mode="json")
         # "name"
-        for i, field in enumerate(fr_template_fields):
+        for field in fr_template_fields:
             value = None
             priority_order = None
             # TODO How to discover Device specific variable
@@ -575,16 +577,19 @@ def generate_feature_template_payload(
                     ):
                         priority_order = get_extra_field(field_value, "priority_order")  # type: ignore
                         value = getattr(template, field_name)
+                        json_dumped_value = json_dumped_template.get(field_name)
                         break
                 if value is None:
                     continue
 
-            # print(field.payload_scheme(value))
-            payload.definition = merge(payload.definition, field.payload_scheme(value, priority_order=priority_order))
+            payload.definition = merge(
+                payload.definition,
+                field.payload_scheme(value, json_dumped_value=json_dumped_value, priority_order=priority_order),
+            )
 
         if debug:
             with open(f"payload_{template.type}.json", "w") as f:
-                f.write(json.dumps(payload.model_dump(by_alias=True), indent=4))
+                f.write(json.dumps(payload.model_dump(by_alias=True, mode="json"), indent=4))
 
         return payload
 

diff --git a/catalystwan/api/templates/bool_str.py b/catalystwan/api/templates/bool_str.py
@@ -0,0 +1,4 @@
+from pydantic import PlainSerializer
+from typing_extensions import Annotated
+
+BoolStr = Annotated[bool, PlainSerializer(lambda x: str(x).lower(), return_type=str, when_used="json-unless-none")]
diff --git a/catalystwan/api/templates/feature_template.py b/catalystwan/api/templates/feature_template.py
@@ -29,7 +29,7 @@ def generate_payload(self, session: ManagerSession) -> str:
             undefined=DebugUndefined,
         )
         template = env.get_template(self.payload_path.name)
-        output = template.render(self.model_dump())
+        output = template.render(self.model_dump(mode="json"))
 
         ast = env.parse(output)
         if meta.find_undeclared_variables(ast):

diff --git a/catalystwan/api/templates/feature_template_field.py b/catalystwan/api/templates/feature_template_field.py
@@ -89,7 +89,7 @@ def data_path(self, output):
 
     # value must be JSON serializable, return JSON serializable dict
     def payload_scheme(
-        self, value: Any = None, help=None, current_path=None, priority_order=None, vip_type=None
+        self, value: Any = None, json_dumped_value: Any = None, priority_order=None, vip_type=None
     ) -> dict:
         output: dict = {}
         rel_output: dict = {}
@@ -119,10 +119,9 @@ def nest_value_in_output(value: Any) -> dict:
                 if self.children:
                     children_output = []
                     for obj in value:  # obj is User, atomic value. Loop every child
+                        obj_json_dump = obj.model_dump(mode="json")
                         child_payload: dict = {}
                         for child in self.children:  # Child in schema
-                            if current_path is None:
-                                current_path = []
                             obj: FeatureTemplate  # type: ignore
                             model_tuple = next(
                                 filter(
@@ -137,14 +136,14 @@ def nest_value_in_output(value: Any) -> dict:
                             )
                             model_field = model_tuple[1]
                             obj_value = getattr(obj, model_tuple[0])
+                            obj_json_value = obj_json_dump.get(model_tuple[0])
                             po = get_extra_field(model_field, "priority_order")
                             vip_type = get_extra_field(model_field, "vip_type")
                             merge(
                                 child_payload,
                                 child.payload_scheme(
                                     obj_value,
-                                    help=output,
-                                    current_path=self.dataPath + [self.key],
+                                    json_dumped_value=obj_json_value,
                                     priority_order=po,
                                     vip_type=vip_type,
                                 ),
@@ -154,7 +153,7 @@ def nest_value_in_output(value: Any) -> dict:
                         children_output.append(child_payload)
                     output["vipValue"] = children_output
                 else:
-                    output["vipValue"] = value
+                    output["vipValue"] = json_dumped_value
             else:
                 if value is None:
                     return {}

diff --git a/catalystwan/api/templates/models/cisco_bfd_model.py b/catalystwan/api/templates/models/cisco_bfd_model.py
@@ -2,10 +2,10 @@
 from pathlib import Path
 from typing import ClassVar, List, Optional
 
-from pydantic import ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field
 
+from catalystwan.api.templates.bool_str import BoolStr
 from catalystwan.api.templates.feature_template import FeatureTemplate
-from catalystwan.utils.pydantic_validators import ConvertBoolToStringModel
 
 DEFAULT_BFD_COLOR_MULTIPLIER = 7
 DEFAULT_BFD_DSCP = 48
@@ -39,18 +39,18 @@ class ColorType(str, Enum):
     PRIVATE6 = "private6"
 
 
-class Color(ConvertBoolToStringModel):
+class Color(BaseModel):
     color: ColorType
     hello_interval: Optional[int] = Field(
         DEFAULT_BFD_HELLO_INTERVAL, json_schema_extra={"vmanage_key": "hello-interval"}
     )
     multiplier: Optional[int] = DEFAULT_BFD_COLOR_MULTIPLIER
-    pmtu_discovery: Optional[bool] = Field(True, json_schema_extra={"vmanage_key": "pmtu-discovery"})
+    pmtu_discovery: Optional[BoolStr] = Field(default=True, json_schema_extra={"vmanage_key": "pmtu-discovery"})
     dscp: Optional[int] = DEFAULT_BFD_DSCP
     model_config = ConfigDict(populate_by_name=True)
 
 
-class CiscoBFDModel(FeatureTemplate, ConvertBoolToStringModel):
+class CiscoBFDModel(FeatureTemplate):
     model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True)
 
     multiplier: Optional[int] = Field(DEFAULT_BFD_MULTIPLIER, json_schema_extra={"data_path": ["app-route"]})

diff --git a/catalystwan/api/templates/models/cisco_bgp_model.py b/catalystwan/api/templates/models/cisco_bgp_model.py
@@ -2,8 +2,9 @@
 from pathlib import Path
 from typing import ClassVar, List, Optional
 
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field
 
+from catalystwan.api.templates.bool_str import BoolStr
 from catalystwan.api.templates.feature_template import FeatureTemplate
 
 
@@ -42,16 +43,10 @@ class AddressFamilyType(str, Enum):
 
 class AggregateAddress(BaseModel):
     prefix: str
-    as_set: Optional[bool] = Field(default=None, json_schema_extra={"vmanage_key": "as-set"})
-    summary_only: Optional[bool] = Field(default=None, json_schema_extra={"vmanage_key": "summary-only"})
+    as_set: Optional[BoolStr] = Field(default=None, json_schema_extra={"vmanage_key": "as-set"})
+    summary_only: Optional[BoolStr] = Field(default=None, json_schema_extra={"vmanage_key": "summary-only"})
     model_config = ConfigDict(populate_by_name=True)
 
-    @field_validator("as_set", "summary_only")
-    @classmethod
-    def cast_to_str(cls, value):
-        if value is not None:
-            return str(value).lower()
-
 
 class Ipv6AggregateAddress(BaseModel):
     prefix: str
@@ -95,20 +90,14 @@ class AddressFamily(BaseModel):
     network: Optional[List[Network]] = None
     ipv6_network: Optional[List[Ipv6Network]] = Field(default=None, json_schema_extra={"vmanage_key": "ipv6-network"})
     paths: Optional[int] = Field(default=None, json_schema_extra={"data_path": ["maximum-paths"]})
-    originate: Optional[bool] = Field(default=None, json_schema_extra={"data_path": ["default-information"]})
+    originate: Optional[BoolStr] = Field(default=None, json_schema_extra={"data_path": ["default-information"]})
     policy_name: Optional[str] = Field(
         default=None, json_schema_extra={"data_path": ["table-map"], "vmanage_key": "name"}
     )
-    filter: Optional[bool] = Field(default=None, json_schema_extra={"data_path": ["table-map"]})
+    filter: Optional[BoolStr] = Field(default=None, json_schema_extra={"data_path": ["table-map"]})
     redistribute: Optional[List[Redistribute]] = None
     model_config = ConfigDict(populate_by_name=True)
 
-    @field_validator("originate", "filter")
-    @classmethod
-    def cast_to_str(cls, value):
-        if value is not None:
-            return str(value).lower()
-
 
 class NeighborFamilyType(str, Enum):
     IPV4_UNICAST = "ipv4-unicast"
@@ -144,21 +133,23 @@ class NeighborAddressFamily(BaseModel):
 class Neighbor(BaseModel):
     address: str
     description: Optional[str] = None
-    shutdown: Optional[bool] = None
+    shutdown: Optional[BoolStr] = None
     remote_as: int = Field(json_schema_extra={"vmanage_key": "remote-as"})
     keepalive: Optional[int] = Field(default=None, json_schema_extra={"data_path": ["timers"]})
     holdtime: Optional[int] = Field(default=None, json_schema_extra={"data_path": ["timers"]})
     if_name: Optional[str] = Field(
         default=None, json_schema_extra={"data_path": ["update-source"], "vmanage_key": "if-name"}
     )
-    next_hop_self: Optional[bool] = Field(default=None, json_schema_extra={"vmanage_key": "next-hop-self"})
-    send_community: Optional[bool] = Field(default=None, json_schema_extra={"vmanage_key": "send-community"})
-    send_ext_community: Optional[bool] = Field(default=None, json_schema_extra={"vmanage_key": "send-ext-community"})
+    next_hop_self: Optional[BoolStr] = Field(default=None, json_schema_extra={"vmanage_key": "next-hop-self"})
+    send_community: Optional[BoolStr] = Field(default=None, json_schema_extra={"vmanage_key": "send-community"})
+    send_ext_community: Optional[BoolStr] = Field(default=None, json_schema_extra={"vmanage_key": "send-ext-community"})
     ebgp_multihop: Optional[int] = Field(default=None, json_schema_extra={"vmanage_key": "ebgp-multihop"})
     password: Optional[str] = None
-    send_label: Optional[bool] = Field(default=None, json_schema_extra={"vmanage_key": "send-label"})
-    send_label_explicit: Optional[bool] = Field(default=None, json_schema_extra={"vmanage_key": "send-label-explicit"})
-    as_override: Optional[bool] = Field(default=None, json_schema_extra={"vmanage_key": "as-override"})
+    send_label: Optional[BoolStr] = Field(default=None, json_schema_extra={"vmanage_key": "send-label"})
+    send_label_explicit: Optional[BoolStr] = Field(
+        default=None, json_schema_extra={"vmanage_key": "send-label-explicit"}
+    )
+    as_override: Optional[BoolStr] = Field(default=None, json_schema_extra={"vmanage_key": "as-override"})
     as_number: Optional[int] = Field(
         default=None, json_schema_extra={"data_path": ["allowas-in"], "vmanage_key": "as-number"}
     )
@@ -167,20 +158,6 @@ class Neighbor(BaseModel):
     )
     model_config = ConfigDict(populate_by_name=True)
 
-    @field_validator(
-        "shutdown",
-        "next_hop_self",
-        "send_community",
-        "send_ext_community",
-        "send_label",
-        "send_label_explicit",
-        "as_override",
-    )
-    @classmethod
-    def cast_to_str(cls, value):
-        if value is not None:
-            return str(value).lower()
-
 
 class IPv6NeighborFamilyType(str, Enum):
     IPV6_UNICAST = "ipv6-unicast"
@@ -203,21 +180,23 @@ class IPv6NeighborAddressFamily(BaseModel):
 class Ipv6Neighbor(BaseModel):
     address: str
     description: Optional[str] = None
-    shutdown: Optional[bool] = None
+    shutdown: Optional[BoolStr] = None
     remote_as: int = Field(default=None, json_schema_extra={"vmanage_key": "remote-as"})
     keepalive: Optional[int] = Field(default=None, json_schema_extra={"data_path": ["timers"]})
     holdtime: Optional[int] = Field(default=None, json_schema_extra={"data_path": ["timers"]})
     if_name: Optional[str] = Field(
         default=None, json_schema_extra={"data_path": ["update-source"], "vmanage_key": "if-name"}
     )
-    next_hop_self: Optional[bool] = Field(False, json_schema_extra={"vmanage_key": "next-hop-self"})
-    send_community: Optional[bool] = Field(True, json_schema_extra={"vmanage_key": "send-community"})
-    send_ext_community: Optional[bool] = Field(True, json_schema_extra={"vmanage_key": "send-ext-community"})
+    next_hop_self: Optional[BoolStr] = Field(default=False, json_schema_extra={"vmanage_key": "next-hop-self"})
+    send_community: Optional[BoolStr] = Field(default=True, json_schema_extra={"vmanage_key": "send-community"})
+    send_ext_community: Optional[BoolStr] = Field(default=True, json_schema_extra={"vmanage_key": "send-ext-community"})
     ebgp_multihop: Optional[int] = Field(1, json_schema_extra={"vmanage_key": "ebgp-multihop"})
     password: Optional[str] = None
-    send_label: Optional[bool] = Field(False, json_schema_extra={"vmanage_key": "send-label"})
-    send_label_explicit: Optional[bool] = Field(False, json_schema_extra={"vmanage_key": "send-label-explicit"})
-    as_override: Optional[bool] = Field(False, json_schema_extra={"vmanage_key": "as-override"})
+    send_label: Optional[BoolStr] = Field(default=False, json_schema_extra={"vmanage_key": "send-label"})
+    send_label_explicit: Optional[BoolStr] = Field(
+        default=False, json_schema_extra={"vmanage_key": "send-label-explicit"}
+    )
+    as_override: Optional[BoolStr] = Field(default=False, json_schema_extra={"vmanage_key": "as-override"})
     as_number: Optional[int] = Field(
         default=None, json_schema_extra={"data_path": ["allowas-in"], "vmanage_key": "as-number"}
     )
@@ -226,26 +205,12 @@ class Ipv6Neighbor(BaseModel):
     )
     model_config = ConfigDict(populate_by_name=True)
 
-    @field_validator(
-        "shutdown",
-        "next_hop_self",
-        "send_community",
-        "send_ext_community",
-        "send_label",
-        "send_label_explicit",
-        "as_override",
-    )
-    @classmethod
-    def cast_to_str(cls, value):
-        if value is not None:
-            return str(value).lower()
-
 
 class CiscoBGPModel(FeatureTemplate):
     model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True)
 
     as_num: Optional[str] = Field(default=None, json_schema_extra={"data_path": ["bgp"], "vmanage_key": "as-num"})
-    shutdown: Optional[bool] = Field(default=None, json_schema_extra={"data_path": ["bgp"]})
+    shutdown: Optional[BoolStr] = Field(default=None, json_schema_extra={"data_path": ["bgp"]})
     router_id: Optional[str] = Field(default=None, json_schema_extra={"data_path": ["bgp"], "vmanage_key": "router-id"})
     propagate_aspath: Optional[bool] = Field(
         default=None, json_schema_extra={"data_path": ["bgp"], "vmanage_key": "propagate-aspath"}
@@ -270,14 +235,14 @@ class CiscoBGPModel(FeatureTemplate):
     always_compare: Optional[bool] = Field(
         default=None, json_schema_extra={"data_path": ["bgp", "best-path", "med"], "vmanage_key": "always-compare"}
     )
-    deterministic: Optional[bool] = Field(default=None, json_schema_extra={"data_path": ["bgp", "best-path", "med"]})
-    missing_as_worst: Optional[bool] = Field(
+    deterministic: Optional[BoolStr] = Field(default=None, json_schema_extra={"data_path": ["bgp", "best-path", "med"]})
+    missing_as_worst: Optional[BoolStr] = Field(
         default=None, json_schema_extra={"data_path": ["bgp", "best-path", "med"], "vmanage_key": "missing-as-worst"}
     )
-    compare_router_id: Optional[bool] = Field(
+    compare_router_id: Optional[BoolStr] = Field(
         default=None, json_schema_extra={"data_path": ["bgp", "best-path"], "vmanage_key": "compare-router-id"}
     )
-    multipath_relax: Optional[bool] = Field(
+    multipath_relax: Optional[BoolStr] = Field(
         default=None, json_schema_extra={"data_path": ["bgp", "best-path", "as-path"], "vmanage_key": "multipath-relax"}
     )
     address_family: Optional[List[AddressFamily]] = Field(
@@ -290,9 +255,3 @@ class CiscoBGPModel(FeatureTemplate):
 
     payload_path: ClassVar[Path] = Path(__file__).parent / "DEPRECATED"
     type: ClassVar[str] = "cisco_bgp"
-
-    @field_validator("shutdown", "deterministic", "missing_as_worst", "compare_router_id", "multipath_relax")
-    @classmethod
-    def cast_to_str(cls, value):
-        if value is not None:
-            return str(value).lower()