fix config_migration

catalystwan/api/configuration_groups/parcel.py

@@ -11,12 +11,14 @@
     model_serializer,
 )
 
+from catalystwan.exceptions import CatalystwanException
+
 T = TypeVar("T")
 
 
 class _ParcelBase(BaseModel):
     model_config = ConfigDict(
-        extra="allow", arbitrary_types_allowed=True, populate_by_name=True,  # json_schema_mode_override="validation"
+        extra="allow", arbitrary_types_allowed=True, populate_by_name=True, json_schema_mode_override="validation"
     )
     parcel_name: str = Field(
         min_length=1,
@@ -33,6 +35,13 @@ class _ParcelBase(BaseModel):
     )
     _parcel_data_key: str = PrivateAttr(default="data")
 
+    @classmethod
+    def _get_parcel_type(cls) -> str:
+        field_info = cls.model_fields.get("type_")
+        if field_info is not None:
+            return str(field_info.default)
+        raise CatalystwanException("Cannot obtain parcel type string")
+
     @model_serializer(mode="wrap")
     def envelope_parcel_data(self, handler: SerializerFunctionWrapHandler) -> Dict[str, Any]:
         """

catalystwan/api/feature_profile_api.py

@@ -18,7 +18,6 @@
     ParcelCreationResponse,
 )
 from catalystwan.models.configuration.feature_profile.sdwan.policy_object import (
-    POLICY_OBJECT_PAYLOAD_ENDPOINT_MAPPING,
     AnyPolicyObjectParcel,
     ApplicationListParcel,
     AppProbeParcel,
@@ -42,8 +41,6 @@
     SecurityZoneListParcel,
     StandardCommunityParcel,
     TlocParcel,
-    URLAllowParcel,
-    URLBlockParcel,
 )
 
 
@@ -200,13 +197,13 @@ def get(self, profile_id: UUID, parcel_type: Type[StandardCommunityParcel]) -> D
     def get(self, profile_id: UUID, parcel_type: Type[TlocParcel]) -> DataSequence[Parcel[Any]]:
         ...
 
-    @overload
-    def get(self, profile_id: UUID, parcel_type: Type[URLAllowParcel]) -> DataSequence[Parcel[Any]]:
-        ...
+    # @overload
+    # def get(self, profile_id: UUID, parcel_type: Type[URLAllowParcel]) -> DataSequence[Parcel[Any]]:
+    #     ...
 
-    @overload
-    def get(self, profile_id: UUID, parcel_type: Type[URLBlockParcel]) -> DataSequence[Parcel[Any]]:
-        ...
+    # @overload
+    # def get(self, profile_id: UUID, parcel_type: Type[URLBlockParcel]) -> DataSequence[Parcel[Any]]:
+    #     ...
 
     # get by id
 
@@ -326,13 +323,13 @@ def get(
     def get(self, profile_id: UUID, parcel_type: Type[TlocParcel], parcel_id: UUID) -> DataSequence[Parcel[Any]]:
         ...
 
-    @overload
-    def get(self, profile_id: UUID, parcel_type: Type[URLAllowParcel], parcel_id: UUID) -> DataSequence[Parcel[Any]]:
-        ...
+    # @overload
+    # def get(self, profile_id: UUID, parcel_type: Type[URLAllowParcel], parcel_id: UUID) -> DataSequence[Parcel[Any]]:
+    #     ...
 
-    @overload
-    def get(self, profile_id: UUID, parcel_type: Type[URLBlockParcel], parcel_id: UUID) -> DataSequence[Parcel[Any]]:
-        ...
+    # @overload
+    # def get(self, profile_id: UUID, parcel_type: Type[URLBlockParcel], parcel_id: UUID) -> DataSequence[Parcel[Any]]:
+    #     ...
 
     def get(
         self,
@@ -344,7 +341,7 @@ def get(
         Get all Policy Objects for selected profile_id and selected type or get one Policy Object given parcel id
         """
 
-        policy_object_list_type = POLICY_OBJECT_PAYLOAD_ENDPOINT_MAPPING[parcel_type]
+        policy_object_list_type = parcel_type._get_parcel_type()
         if not parcel_id:
             return self.endpoint.get_all(profile_id=profile_id, policy_object_list_type=policy_object_list_type)
         parcel = self.endpoint.get_by_id(
@@ -357,7 +354,7 @@ def create(self, profile_id: UUID, payload: AnyPolicyObjectParcel) -> ParcelCrea
         Create Policy Object for selected profile_id based on payload type
         """
 
-        policy_object_list_type = POLICY_OBJECT_PAYLOAD_ENDPOINT_MAPPING[type(payload)]
+        policy_object_list_type = payload._get_parcel_type()
         return self.endpoint.create(
             profile_id=profile_id, policy_object_list_type=policy_object_list_type, payload=payload
         )
@@ -367,7 +364,7 @@ def update(self, profile_id: UUID, payload: AnyPolicyObjectParcel, list_object_i
         Update Policy Object for selected profile_id based on payload type
         """
 
-        policy_type = POLICY_OBJECT_PAYLOAD_ENDPOINT_MAPPING[type(payload)]
+        policy_type = payload._get_parcel_type()
         return self.endpoint.update(
             profile_id=profile_id, policy_object_list_type=policy_type, list_object_id=list_object_id, payload=payload
         )
@@ -460,20 +457,20 @@ def delete(self, profile_id: UUID, parcel_type: Type[StandardCommunityParcel], l
     def delete(self, profile_id: UUID, parcel_type: Type[TlocParcel], list_object_id: UUID) -> None:
         ...
 
-    @overload
-    def delete(self, profile_id: UUID, parcel_type: Type[URLAllowParcel], list_object_id: UUID) -> None:
-        ...
+    # @overload
+    # def delete(self, profile_id: UUID, parcel_type: Type[URLAllowParcel], list_object_id: UUID) -> None:
+    #     ...
 
-    @overload
-    def delete(self, profile_id: UUID, parcel_type: Type[URLBlockParcel], list_object_id: UUID) -> None:
-        ...
+    # @overload
+    # def delete(self, profile_id: UUID, parcel_type: Type[URLBlockParcel], list_object_id: UUID) -> None:
+    #     ...
 
     def delete(self, profile_id: UUID, parcel_type: Type[AnyPolicyObjectParcel], list_object_id: UUID) -> None:
         """
         Delete Policy Object for selected profile_id based on payload type
         """
 
-        policy_object_list_type = POLICY_OBJECT_PAYLOAD_ENDPOINT_MAPPING[parcel_type]
+        policy_object_list_type = parcel_type._get_parcel_type()
         return self.endpoint.delete(
             profile_id=profile_id, policy_object_list_type=policy_object_list_type, list_object_id=list_object_id
         )

catalystwan/models/configuration/feature_profile/sdwan/policy_object/__init__.py

@@ -1,4 +1,4 @@
-from typing import List, Mapping, Union
+from typing import List, Union
 
 from pydantic import Field
 from typing_extensions import Annotated
@@ -42,7 +42,7 @@
 
 AnyPolicyObjectParcel = Annotated[
     Union[
-        AnyURLParcel,
+        # AnyURLParcel,
         ApplicationListParcel,
         AppProbeParcel,
         ColorParcel,
@@ -70,34 +70,6 @@
     Field(discriminator="type_"),
 ]
 
-POLICY_OBJECT_PAYLOAD_ENDPOINT_MAPPING: Mapping[type, str] = {
-    AppProbeParcel: "app-probe",
-    ApplicationListParcel: "app-list",
-    ColorParcel: "color",
-    DataPrefixParcel: "data-prefix",
-    ExpandedCommunityParcel: "expanded-community",
-    FowardingClassParcel: "class",
-    IPv6DataPrefixParcel: "data-ipv6-prefix",
-    IPv6PrefixListParcel: "ipv6-prefix",
-    PrefixListParcel: "prefix",
-    PolicierParcel: "policer",
-    PreferredColorGroupParcel: "preferred-color-group",
-    SLAClassParcel: "sla-class",
-    TlocParcel: "tloc",
-    StandardCommunityParcel: "standard-community",
-    LocalDomainParcel: "security-localdomain",
-    FQDNDomainParcel: "security-fqdn",
-    IPSSignatureParcel: "security-ipssignature",
-    URLAllowParcel: "security-urllist",
-    URLBlockParcel: "security-urllist",
-    SecurityPortParcel: "security-port",
-    ProtocolListParcel: "security-protocolname",
-    GeoLocationListParcel: "security-geolocation",
-    SecurityZoneListParcel: "security-zone",
-    SecurityApplicationListParcel: "security-localapp",
-    SecurityDataPrefixParcel: "security-data-ip-prefix",
-}
-
 __all__ = (
     "AnyPolicyObjectParcel",
     "ApplicationFamilyListEntry",

catalystwan/models/policy/lists_entries.py

@@ -7,18 +7,21 @@
 from catalystwan.models.common import InterfaceType, TLOCColor, check_fields_exclusive
 
 
-def check_jitter_ms(jitter_str: str) -> str:
-    assert 1 <= int(jitter_str) <= 1000
+def check_jitter_ms(jitter_str: Optional[str]) -> Optional[str]:
+    if jitter_str is not None:
+        assert 1 <= int(jitter_str) <= 1000
     return jitter_str
 
 
-def check_latency_ms(latency_str: str) -> str:
-    assert 1 <= int(latency_str) <= 1000
+def check_latency_ms(latency_str: Optional[str]) -> Optional[str]:
+    if latency_str is not None:
+        assert 1 <= int(latency_str) <= 1000
     return latency_str
 
 
-def check_loss_percent(loss_str: str) -> str:
-    assert 0 <= int(loss_str) <= 100
+def check_loss_percent(loss_str: Optional[str]) -> Optional[str]:
+    if loss_str is not None:
+        assert 0 <= int(loss_str) <= 100
     return loss_str
 
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "catalystwan"
-version = "0.31.0dev2"
+version = "0.31.0dev3"
 description = "Cisco Catalyst WAN SDK for Python"
 authors = ["kagorski <[email protected]>"]
 readme = "README.md"