remove pydantic v1 models (#10)

* remove pydantic v1 models

* fix problems

* bump minor version

README.md

@@ -288,13 +288,13 @@ api = session.api.tenant_management
 tenants = [
     Tenant(
         name="tenant1",
-        orgName="CiscoDevNet",
-        subDomain="alpha.bravo.net",
+        org_name="CiscoDevNet",
+        subdomain="alpha.bravo.net",
         desc="This is tenant for unit tests",
-        edgeConnectorEnable=True,
-        edgeConnectorSystemIp="172.16.255.81",
-        edgeConnectorTunnelInterfaceName="GigabitEthernet1",
-        wanEdgeForecast=1,
+        edge_connector_enable=True,
+        edge_connector_system_ip="172.16.255.81",
+        edge_connector_tunnel_interface_name="GigabitEthernet1",
+        wan_edge_forecast=1,
     )
 ]
 create_task = api.create(tenants)

catalystwan/api/tenant_management_api.py

@@ -74,7 +74,7 @@ def delete(self, tenant_id_list: List[str], password: Optional[str] = None) -> T
         """
         if password is None:
             password = self.session.password
-        delete_request = TenantBulkDeleteRequest(tenantIdList=tenant_id_list, password=password)
+        delete_request = TenantBulkDeleteRequest(tenant_id_list=tenant_id_list, password=password)
         task_id = self._endpoints.delete_tenant_async_bulk(delete_request).id
         return Task(self.session, task_id)
 
@@ -105,7 +105,7 @@ def update_vsmart_placement(self, tenant_id: str, src_vsmart_uuid: str, dst_vsma
         self._endpoints.update_tenant_vsmart_placement(
             tenant_id=tenant_id,
             vsmart_placement_update_request=vSmartPlacementUpdateRequest(
-                srcvSmartUuid=src_vsmart_uuid, destvSmartUuid=dst_vsmart_uuid
+                src_vsmart_uuid=src_vsmart_uuid, dest_vsmart_uuid=dst_vsmart_uuid
             ),
         )
 

catalystwan/api/tenant_migration_api.py

@@ -101,7 +101,7 @@ def store_token(self, migration_id: str, download_path: Path):
             migration_id (str): migration identifier (it is obtained after import tenant task is finished)
             download_path (Path): full download path containing a filename eg.: Path("/home/user/import-token.txt")
         """
-        params = MigrationTokenQueryParams(migrationId=migration_id)
+        params = MigrationTokenQueryParams(migration_id=migration_id)
         token = self.session.endpoints.tenant_migration.get_migration_token(params)
         with open(download_path, "w") as file:
             file.write(token)

catalystwan/dataclasses.py

@@ -4,7 +4,7 @@
 from typing import List, Optional
 
 from attr import define, field  # type: ignore
-from pydantic.v1 import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 from catalystwan.exceptions import RetrieveIntervalOutOfRange
 from catalystwan.utils.alarm_status import Severity
@@ -504,16 +504,17 @@ class SoftwareInstallTimeout(DataclassBase):
 
 
 class FeatureTemplatesTypes(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
     parent: str
     default: str
-    display_name: str = Field(alias="displayName")
+    display_name: str = Field(serialization_alias="displayName", validation_alias="displayName")
     name: str
-    type_class: str = Field(alias="typeClass")
+    type_class: str = Field(serialization_alias="typeClass", validation_alias="typeClass")
     description: str
     write_permission: bool
     read_permission: bool
-    helper_type: List[str] = Field(default=[], alias="helperType")
-    device_models: List[dict] = Field(default=[], alias="deviceModels")
+    helper_type: List[str] = Field(default=[], serialization_alias="helperType", validation_alias="helperType")
+    device_models: List[dict] = Field(default=[], serialization_alias="deviceModels", validation_alias="deviceModels")
 
 
 @define

catalystwan/endpoints/__init__.py

@@ -61,8 +61,7 @@
 
 from packaging.specifiers import SpecifierSet  # type: ignore
 from packaging.version import Version  # type: ignore
-from pydantic import BaseModel as BaseModelV2
-from pydantic.v1 import BaseModel as BaseModelV1
+from pydantic import BaseModel
 from typing_extensions import Annotated, get_args, get_origin
 
 from catalystwan.abstractions import APIEndpointClient, APIEndpointClientResponse
@@ -82,12 +81,10 @@ def prepared(self) -> PreparedPayload:
 
 
 JSON = Union[str, int, float, bool, None, Dict[str, "JSON"], List["JSON"]]
-ModelPayloadType = Union[BaseModelV1, BaseModelV2, Sequence[BaseModelV1], Sequence[BaseModelV2]]
+ModelPayloadType = Union[BaseModel, Sequence[BaseModel]]
 PayloadType = Union[None, JSON, str, bytes, dict, ModelPayloadType, CustomPayloadType]
-ReturnType = Union[
-    None, JSON, bytes, str, dict, BaseModelV1, BaseModelV2, DataSequence[BaseModelV1], DataSequence[BaseModelV2]
-]
-RequestParamsType = Union[Dict[str, str], BaseModelV1, BaseModelV2]
+ReturnType = Union[None, JSON, bytes, str, dict, BaseModel, DataSequence[BaseModel]]
+RequestParamsType = Union[Dict[str, str], BaseModel]
 
 
 @dataclass
@@ -119,12 +116,12 @@ def model_union(cls, models: Sequence[type]) -> TypeSpecifier:
 
     @classmethod
     def resolve_nested_base_model_unions(
-        cls, annotation: Any, models_types: List[Union[Type[BaseModelV1], Type[BaseModelV2]]]
-    ) -> List[Union[Type[BaseModelV1], Type[BaseModelV2]]]:
+        cls, annotation: Any, models_types: List[Type[BaseModel]]
+    ) -> List[Type[BaseModel]]:
         type_origin = get_origin(annotation)
         if isclass(annotation):
             try:
-                if issubclass(annotation, (BaseModelV1, BaseModelV2)):
+                if issubclass(annotation, BaseModel):
                     return [annotation]
                 raise APIEndpointError(f"Expected: {PayloadType}")
             except TypeError:
@@ -135,7 +132,7 @@ def resolve_nested_base_model_unions(
                 if (len(annotated_origin) >= 1) and get_origin(annotated_origin[0]) == Union:
                     type_args = get_args(annotated_origin[0])
                     if all(isclass(t) for t in type_args) and all(
-                        issubclass(t, (BaseModelV1, BaseModelV2)) for t in type_args
+                        issubclass(t, BaseModel) for t in type_args
                     ):
                         models_types.extend(list(type_args))
                         return models_types
@@ -148,7 +145,7 @@ def resolve_nested_base_model_unions(
         # Check if Union[PayloadModelType, ...], only unions of pydantic models allowed
         elif type_origin == Union:
             type_args = get_args(annotation)
-            if all(isclass(t) for t in type_args) and all(issubclass(t, (BaseModelV1, BaseModelV2)) for t in type_args):
+            if all(isclass(t) for t in type_args) and all(issubclass(t, BaseModel) for t in type_args):
                 models_types.extend(list(type_args))
                 return models_types
             else:
@@ -212,7 +209,7 @@ def _prepare_payload(cls, payload: PayloadType, force_json: bool = False) -> Pre
             return PreparedPayload(data=json.dumps(payload), headers={"content-type": "application/json"})
         if isinstance(payload, (str, bytes)):
             return PreparedPayload(data=payload)
-        elif isinstance(payload, (BaseModelV1, BaseModelV2)):
+        elif isinstance(payload, (BaseModel)):
             return cls._prepare_basemodel_payload(payload)
         elif isinstance(payload, Sequence) and not isinstance(payload, (str, bytes)):
             return cls._prepare_sequence_payload(payload)  # type: ignore[arg-type]
@@ -223,34 +220,25 @@ def _prepare_payload(cls, payload: PayloadType, force_json: bool = False) -> Pre
             raise APIRequestPayloadTypeError(payload)
 
     @classmethod
-    def _prepare_basemodel_payload(cls, payload: Union[BaseModelV1, BaseModelV2]) -> PreparedPayload:
+    def _prepare_basemodel_payload(cls, payload: BaseModel) -> PreparedPayload:
         """Helper method to prepare BaseModel instance for sending"""
-        if isinstance(payload, BaseModelV1):
-            return PreparedPayload(
-                data=payload.json(exclude_none=True, by_alias=True), headers={"content-type": "application/json"}
-            )
         return PreparedPayload(
             data=payload.model_dump_json(exclude_none=True, by_alias=True), headers={"content-type": "application/json"}
         )
 
     @classmethod
-    def _prepare_sequence_payload(cls, payload: Iterable[Union[BaseModelV1, BaseModelV2]]) -> PreparedPayload:
+    def _prepare_sequence_payload(cls, payload: Iterable[BaseModel]) -> PreparedPayload:
         """Helper method to prepare sequences for sending"""
         items = []
         for item in payload:
-            if isinstance(item, BaseModelV1):
-                items.append(item.dict(exclude_none=True, by_alias=True))
-            elif isinstance(item, BaseModelV2):
-                items.append(item.model_dump(exclude_none=True, by_alias=True))
+            items.append(item.model_dump(exclude_none=True, by_alias=True))
         data = json.dumps(items)
         return PreparedPayload(data=data, headers={"content-type": "application/json"})
 
     @classmethod
     def _prepare_params(cls, params: RequestParamsType) -> Dict[str, Any]:
         """Helper method to prepare params for sending"""
-        if isinstance(params, BaseModelV1):
-            return params.dict(exclude_none=True, by_alias=True)
-        elif isinstance(params, BaseModelV2):
+        if isinstance(params, BaseModel):
             return params.model_dump(exclude_none=True, by_alias=True)
         return params
 
@@ -433,13 +421,13 @@ def specify_return_type(self) -> TypeSpecifier:
                 (type_args := get_args(annotation))
                 and (len(type_args) == 1)
                 and isclass(type_args[0])
-                and issubclass(type_args[0], (BaseModelV1, BaseModelV2))
+                and issubclass(type_args[0], BaseModel)
             ):
                 return TypeSpecifier(True, DataSequence, type_args[0])
             raise APIEndpointError(f"Expected: {ReturnType} but return type {annotation}")
         elif isclass(annotation):
             try:
-                if issubclass(annotation, (bytes, str, dict, BinaryIO, (BaseModelV1, BaseModelV2))):
+                if issubclass(annotation, (bytes, str, dict, BinaryIO, BaseModel)):
                     return TypeSpecifier(True, None, annotation)
                 raise APIEndpointError(f"Expected: {ReturnType} but return type {annotation}")
             except TypeError:
@@ -479,7 +467,7 @@ def specify_payload_type(self) -> TypeSpecifier:
 
         # Check if regular class
         if isclass(annotation):
-            if issubclass(annotation, (bytes, str, dict, BinaryIO, BaseModelV1, BaseModelV2, CustomPayloadType)):
+            if issubclass(annotation, (bytes, str, dict, BinaryIO, BaseModel, CustomPayloadType)):
                 return TypeSpecifier(True, None, annotation, None, False, is_optional)
             else:
                 raise APIEndpointError(f"'payload' param must be annotated with supported type: {PayloadType}")
@@ -492,7 +480,7 @@ def specify_payload_type(self) -> TypeSpecifier:
                     (type_args := get_args(annotation))
                     and (len(type_args) == 1)
                     and isclass(type_args[0])
-                    and issubclass(type_args[0], (BaseModelV1, BaseModelV2))
+                    and issubclass(type_args[0], BaseModel)
                 ):
                     return TypeSpecifier(True, type_origin, type_args[0], None, False, is_optional)
             else:
@@ -509,10 +497,7 @@ def check_params(self):
         parameters = self.sig.parameters
 
         if params_param := parameters.get("params"):
-            if not (
-                isclass(params_param.annotation)
-                and issubclass(params_param.annotation, (BaseModelV1, BaseModelV2, Dict))
-            ):
+            if not (isclass(params_param.annotation) and issubclass(params_param.annotation, (BaseModel, Dict))):
                 raise APIEndpointError(f"'params' param must be annotated with supported type: {RequestParamsType}")
 
         general_purpose_arg_names = {
@@ -603,7 +588,7 @@ def wrapper(*args, **kwargs):
                     return full_json
                 if self.return_spec.payload_type is None:
                     pass
-                elif issubclass(self.return_spec.payload_type, (BaseModelV1, BaseModelV2)):
+                elif issubclass(self.return_spec.payload_type, BaseModel):
                     if self.return_spec.sequence_type == DataSequence:
                         return response.dataseq(self.return_spec.payload_type, self.resp_json_key)
                     else:

catalystwan/endpoints/certificate_management_vmanage.py

@@ -3,21 +3,22 @@
 # mypy: disable-error-code="empty-body"
 import datetime
 
-from pydantic.v1 import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 from catalystwan.endpoints import APIEndpoints, get
 
 
 class WebServerCertificateInfo(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
     org_unit: str
     org: str
     location: str
     state: str
     country: str
     company_name: str
-    not_before: datetime.datetime = Field(alias="notBefore")
-    not_after: datetime.datetime = Field(alias="notAfter")
-    certificate_details: str = Field(alias="certificateDetails")
+    not_before: datetime.datetime = Field(serialization_alias="notBefore", validation_alias="notBefore")
+    not_after: datetime.datetime = Field(serialization_alias="notAfter", validation_alias="notAfter")
+    certificate_details: str = Field(serialization_alias="certificateDetails", validation_alias="certificateDetails")
     validity: str
 
 

catalystwan/endpoints/client.py

@@ -5,7 +5,7 @@
 from typing import Any, List, Optional
 
 from packaging.version import Version  # type: ignore
-from pydantic.v1 import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 from catalystwan.endpoints import APIEndpoints, get
 
@@ -21,46 +21,65 @@ def validate(cls, value):
 
 
 class ServerInfo(BaseModel):
-    server: Optional[str]
-    tenancy_mode: Optional[str] = Field(alias="tenancyMode")
-    user_mode: Optional[str] = Field(alias="userMode")
-    vsession_id: Optional[str] = Field(alias="VSessionId")
-    is_saml_user: Optional[bool] = Field(alias="isSamlUser")
-    is_rbac_vpn_user: Optional[bool] = Field(alias="isRbacVpnUser")
+    model_config = ConfigDict(populate_by_name=True)
+    server: Optional[str] = None
+    tenancy_mode: Optional[str] = Field(default=None, serialization_alias="tenancyMode", validation_alias="tenancyMode")
+    user_mode: Optional[str] = Field(default=None, serialization_alias="userMode", validation_alias="userMode")
+    vsession_id: Optional[str] = Field(default=None, serialization_alias="VSessionId", validation_alias="VSessionId")
+    is_saml_user: Optional[bool] = Field(default=None, serialization_alias="isSamlUser", validation_alias="isSamlUser")
+    is_rbac_vpn_user: Optional[bool] = Field(
+        default=None, serialization_alias="isRbacVpnUser", validation_alias="isRbacVpnUser"
+    )
     vpns: List[Any] = []
-    csrf_token: Optional[str] = Field(alias="CSRFToken")
-    provider_domain: Optional[str] = Field(alias="providerDomain")
-    tenant_id: Optional[str] = Field(alias="tenantId")
-    provider_id: Optional[str] = Field(alias="providerId")
-    view_mode: Optional[str] = Field(alias="viewMode")
+    csrf_token: Optional[str] = Field(default=None, serialization_alias="CSRFToken", validation_alias="CSRFToken")
+    provider_domain: Optional[str] = Field(
+        default=None, serialization_alias="providerDomain", validation_alias="providerDomain"
+    )
+    tenant_id: Optional[str] = Field(default=None, serialization_alias="tenantId", validation_alias="tenantId")
+    provider_id: Optional[str] = Field(default=None, serialization_alias="providerId", validation_alias="providerId")
+    view_mode: Optional[str] = Field(default=None, serialization_alias="viewMode", validation_alias="viewMode")
     capabilities: List[str] = []
-    user: Optional[str]
-    description: Optional[str]
-    locale: Optional[str]
+    user: Optional[str] = None
+    description: Optional[str] = None
+    locale: Optional[str] = None
     roles: List[str] = []
-    external_user: Optional[bool] = Field(alias="externalUser")
-    platform_version: str = Field(default="", alias="platformVersion")
-    general_template: Optional[bool] = Field(alias="generalTemplate")
-    disable_full_config_push: Optional[bool] = Field(alias="disableFullConfigPush")
-    enable_server_events: Optional[bool] = Field(alias="enableServerEvents")
-    cloudx: Optional[str]
-    reverseproxy: Optional[str]
-    vmanage_mode: Optional[str] = Field(alias="vmanageMode")
+    external_user: Optional[bool] = Field(
+        default=None, serialization_alias="externalUser", validation_alias="externalUser"
+    )
+    platform_version: str = Field(default="", serialization_alias="platformVersion", validation_alias="platformVersion")
+    general_template: Optional[bool] = Field(
+        default=None, serialization_alias="generalTemplate", validation_alias="generalTemplate"
+    )
+    disable_full_config_push: Optional[bool] = Field(
+        default=None, serialization_alias="disableFullConfigPush", validation_alias="disableFullConfigPush"
+    )
+    enable_server_events: Optional[bool] = Field(
+        default=None, serialization_alias="enableServerEvents", validation_alias="enableServerEvents"
+    )
+    cloudx: Optional[str] = None
+    reverseproxy: Optional[str] = None
+    vmanage_mode: Optional[str] = Field(default=None, serialization_alias="vmanageMode", validation_alias="vmanageMode")
 
 
 class AboutInfo(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
     title: Optional[str]
     version: str = Field(default="")
-    application_version: Optional[str] = Field(alias="applicationVersion")
-    application_server: Optional[str] = Field(alias="applicationServer")
+    application_version: Optional[str] = Field(
+        serialization_alias="applicationVersion", validation_alias="applicationVersion"
+    )
+    application_server: Optional[str] = Field(
+        serialization_alias="applicationServer", validation_alias="applicationServer"
+    )
     copyright: Optional[str]
     time: Optional[datetime]
-    time_zone: Optional[str] = Field(alias="timeZone")
+    time_zone: Optional[str] = Field(serialization_alias="timeZone", validation_alias="timeZone")
     logo: Optional[str]
 
 
 class ServerReady(BaseModel):
-    is_server_ready: bool = Field(alias="isServerReady")
+    model_config = ConfigDict(populate_by_name=True)
+    is_server_ready: bool = Field(serialization_alias="isServerReady", validation_alias="isServerReady")
 
 
 class Client(APIEndpoints):

catalystwan/endpoints/cluster_management.py

@@ -1,23 +1,18 @@
 # Copyright 2023 Cisco Systems, Inc. and its affiliates
 
 # mypy: disable-error-code="empty-body"
+from typing import Literal, Optional
 
-from enum import Enum
-from typing import Optional
-
-from pydantic.v1 import BaseModel
+from pydantic import BaseModel
 
 from catalystwan.endpoints import APIEndpoints, get
 from catalystwan.typed_list import DataSequence
 
-
-class TenancyModeEnum(str, Enum):
-    st = "SingleTenant"
-    mt = "MultiTenant"
+TenancyModes = Literal["SingleTenant", "MultiTenant"]
 
 
 class TenancyMode(BaseModel):
-    mode: TenancyModeEnum
+    mode: TenancyModes
     deploymentmode: str
     domain: Optional[str] = None
     clusterid: Optional[str] = None