Migrate to pydantic 2 (#291)

* Upgrade & fix models & tests

* address deprecated pydantic warnings

* remove print statments, uncomment test

* fix optional -> None fields

requirements.txt

@@ -2,7 +2,7 @@
 pytest==7.4.2
 responses==0.22.0
 coverage
-pydantic>=1.10,==1.*
+pydantic==2.5.2
 
 bump2version
 build

setup.py

@@ -25,7 +25,7 @@
         "requests>=2.4.2",
         "pytz>=2018.5",
         "Deprecated",
-        "pydantic>=1.10,==1.*",
+        "pydantic>=2.5.2",
     ],
     python_requires=">=3.8",
     tests_require=["cryptography>=2.3.1"],

src/vonage/ncco_builder/connect_endpoints.py

@@ -1,49 +1,62 @@
-from pydantic import BaseModel, HttpUrl, AnyUrl, Field, constr
-from typing import Optional, Dict
+from pydantic import BaseModel, HttpUrl, AnyUrl, constr, field_serializer
+from typing import Dict
 from typing_extensions import Literal
 
 
 class ConnectEndpoints:
     class Endpoint(BaseModel):
-        type: str = None
+        type: Literal['phone', 'app', 'websocket', 'sip', 'vbc'] = None
 
     class PhoneEndpoint(Endpoint):
-        type = Field('phone', const=True)
-        number: constr(regex=r'^[1-9]\d{6,14}$')
-        dtmfAnswer: Optional[constr(regex='^[0-9*#p]+$')]
-        onAnswer: Optional[Dict[str, HttpUrl]]
+        type: Literal['phone'] = 'phone'
+
+        number: constr(pattern=r'^[1-9]\d{6,14}$')
+        dtmfAnswer: constr(pattern='^[0-9*#p]+$') = None
+        onAnswer: Dict[str, HttpUrl] = None
+
+        @field_serializer('onAnswer')
+        def serialize_dt(self, oa: Dict[str, HttpUrl], _info):
+            if oa is None:
+                return oa
+
+            return {k: str(v) for k, v in oa.items()}
 
     class AppEndpoint(Endpoint):
-        type = Field('app', const=True)
+        type: Literal['app'] = 'app'
         user: str
 
     class WebsocketEndpoint(Endpoint):
-        type = Field('websocket', const=True)
+        type: Literal['websocket'] = 'websocket'
+
         uri: AnyUrl
         contentType: Literal['audio/l16;rate=16000', 'audio/l16;rate=8000']
-        headers: Optional[dict]
+        headers: dict = None
+
+        @field_serializer('uri')
+        def serialize_uri(self, uri: AnyUrl, _info):
+            return str(uri)
 
     class SipEndpoint(Endpoint):
-        type = Field('sip', const=True)
+        type: Literal['sip'] = 'sip'
         uri: str
-        headers: Optional[dict]
+        headers: dict = None
 
     class VbcEndpoint(Endpoint):
-        type = Field('vbc', const=True)
+        type: Literal['vbc'] = 'vbc'
         extension: str
 
     @classmethod
     def create_endpoint_model_from_dict(cls, d) -> Endpoint:
         if d['type'] == 'phone':
-            return cls.PhoneEndpoint.parse_obj(d)
+            return cls.PhoneEndpoint.model_validate(d)
         elif d['type'] == 'app':
-            return cls.AppEndpoint.parse_obj(d)
+            return cls.AppEndpoint.model_validate(d)
         elif d['type'] == 'websocket':
-            return cls.WebsocketEndpoint.parse_obj(d)
+            return cls.WebsocketEndpoint.model_validate(d)
         elif d['type'] == 'sip':
-            return cls.WebsocketEndpoint.parse_obj(d)
+            return cls.WebsocketEndpoint.model_validate(d)
         elif d['type'] == 'vbc':
-            return cls.WebsocketEndpoint.parse_obj(d)
+            return cls.WebsocketEndpoint.model_validate(d)
         else:
             raise ValueError(
                 'Invalid "type" specified for endpoint object. Cannot create a ConnectEndpoints.Endpoint model.'

src/vonage/ncco_builder/input_types.py

@@ -1,26 +1,26 @@
 from pydantic import BaseModel, confloat, conint
-from typing import Optional, List
+from typing import List
 
 
 class InputTypes:
     class Dtmf(BaseModel):
-        timeOut: Optional[conint(ge=0, le=10)]
-        maxDigits: Optional[conint(ge=1, le=20)]
-        submitOnHash: Optional[bool]
+        timeOut: conint(ge=0, le=10) = None
+        maxDigits: conint(ge=1, le=20) = None
+        submitOnHash: bool = None
 
     class Speech(BaseModel):
-        uuid: Optional[str]
-        endOnSilence: Optional[confloat(ge=0.4, le=10.0)]
-        language: Optional[str]
-        context: Optional[List[str]]
-        startTimeout: Optional[conint(ge=1, le=60)]
-        maxDuration: Optional[conint(ge=1, le=60)]
-        saveAudio: Optional[bool]
+        uuid: str = None
+        endOnSilence: confloat(ge=0.4, le=10.0) = None
+        language: str = None
+        context: List[str] = None
+        startTimeout: conint(ge=1, le=60) = None
+        maxDuration: conint(ge=1, le=60) = None
+        saveAudio: bool = None
 
     @classmethod
     def create_dtmf_model(cls, dict) -> Dtmf:
-        return cls.Dtmf.parse_obj(dict)
+        return cls.Dtmf.model_validate(dict)
 
     @classmethod
     def create_speech_model(cls, dict) -> Speech:
-        return cls.Speech.parse_obj(dict)
+        return cls.Speech.model_validate(dict)