Add ncco builder (#236)

* initial Ncco class creation

* adding optional fields, changing structure for namespacing reasons

* added pydantic as dependency

* using List type from typing module

* adding and testing Notify and Talk actions, refactoring build_ncco method

* more endpoints and tests

* added connect_endpoints.py, Endpoints types, refactored into separate modules, testing connect action and endpoint models

* placeholder validator test

* changing NCCO builder test and module structure, add Connect and Stream endpoints, starting Input endpoint

* added Input action and submodels, testing Input action, adding pay and submodels

* renaming to fix validator conflict

* adding pay prompt actions and errors, pay action testing

* adding PayPrompts tests, adding type hints

* finished Pay action, testing Ncco.build_ncco method

* removing custom URL types as they cause problems, testing NCCO builder

* adding full ncco builder test

* added voice test using ncco builder

Author: maxkahan

README.md

@@ -16,6 +16,7 @@ need a Vonage account. Sign up [for free at vonage.com][signup].
 - [SMS API](#sms-api)
 - [Messages API](#messages-api)
 - [Voice API](#voice-api)
+- [NCCO Builder](#ncco-builder)
 - [Verify API](#verify-api)
 - [Number Insight API](#number-insight-api)
 - [Number Management API](#number-management-api)
@@ -337,6 +338,65 @@ client.voice.send_dtmf(response['uuid'], digits='1234')
 response = client.get_recording(RECORDING_URL)
 ```
 
+## NCCO Builder
+
+The SDK contains a builder to help you create Call Control Objects (NCCOs) for use with the Vonage Voice API.
+
+For more information, [check the full NCCO reference documentation on the Vonage website](https://developer.vonage.com/voice/voice-api/ncco-reference).
+
+An NCCO is a list of "Actions": steps to be followed when a call is initiated or received.
+
+Use the builder to construct valid NCCO actions, which are modelled in the SDK as [Pydantic](https://docs.pydantic.dev) models, and build them into an NCCO. The NCCO actions supported by the builder are:
+
+* Record
+* Conversation
+* Connect
+* Talk
+* Stream
+* Input
+* Notify
+* Pay
+
+### Construct actions
+
+```python
+record = Ncco.Record(eventUrl=['https://example.com'])
+talk = Ncco.Talk(text='Hello from Vonage!', bargeIn=True, loop=5, premium=True)
+```
+
+The Connect action has each valid endpoint type (phone, application, WebSocket, SIP and VBC) specified as a Pydantic model so these can be validated, though it is also possible to pass in a dict with the endpoint properties directly into the `Ncco.Connect` object.
+
+This example shows a Connect action created with an endpoint object.
+
+```python
+phone = ConnectEndpoints.PhoneEndpoint(
+        number='447000000000',
+        dtmfAnswer='1p2p3p#**903#',
+    )
+connect = Ncco.Connect(endpoint=phone, eventUrl=['https://example.com/events'], from_='447000000000')
+```
+
+This example shows a different Connect action, created with a dictionary.
+
+```python
+connect = Ncco.Connect(endpoint={'type': 'phone', 'number': '447000000000', 'dtmfAnswer': '2p02p'}, randomFromNumber=True)
+```
+
+### Build into an NCCO
+
+Create an NCCO from the actions with the `Ncco.build_ncco` method. This will be returned as a list of dicts representing each action and can be used in calls to the Voice API.
+
+```python
+ncco = Ncco.build_ncco(record, connect, talk)
+
+response = client.voice.create_call({
+    'to': [{'type': 'phone', 'number': TO_NUMBER}],
+    'from': {'type': 'phone', 'number': VONAGE_NUMBER},
+    'ncco': ncco
+})
+
+pprint(response)
+```
 
 ## Verify API
 

requirements.txt

@@ -2,6 +2,7 @@
 pytest==7.2.0
 responses==0.22.0
 coverage
+pydantic
 
 bump2version
 build

setup.py

@@ -27,6 +27,7 @@
         "PyJWT[crypto]>=1.6.4",
         "pytz>=2018.5",
         "Deprecated",
+        "pydantic>=1.10.2",
     ],
     python_requires=">=3.7",
     tests_require=["cryptography>=2.3.1"],

src/vonage/client.py

@@ -4,6 +4,7 @@
 from .application import ApplicationV2, Application
 from .errors import *
 from .messages import Messages
+from .ncco_builder.ncco import Ncco, ConnectEndpoints, InputTypes, PayPrompts
 from .number_insight import NumberInsight
 from .numbers import Numbers
 from .redact import Redact
@@ -37,6 +38,7 @@
 
 logger = logging.getLogger("vonage")
 
+
 class Client:
     """
     Create a Client object to start making calls to Vonage/Nexmo APIs.
@@ -78,10 +80,10 @@ def __init__(
         private_key=None,
         app_name=None,
         app_version=None,
-        timeout=None, 
-        pool_connections=10, 
-        pool_maxsize=10, 
-        max_retries=3
+        timeout=None,
+        pool_connections=10,
+        pool_maxsize=10,
+        max_retries=3,
     ):
         self.api_key = key or os.environ.get("VONAGE_API_KEY", None)
         self.api_secret = secret or os.environ.get("VONAGE_API_SECRET", None)
@@ -126,9 +128,7 @@ def __init__(
         self.timeout = timeout
         self.session = Session()
         self.adapter = HTTPAdapter(
-            pool_connections=pool_connections, 
-            pool_maxsize=pool_maxsize, 
-            max_retries=max_retries
+            pool_connections=pool_connections, pool_maxsize=pool_maxsize, max_retries=max_retries
         )
         self.session.mount("https://", self.adapter)
 
@@ -156,9 +156,7 @@ def check_signature(self, params):
 
     def signature(self, params):
         if self.signature_method:
-            hasher = hmac.new(
-                self.signature_secret.encode(), digestmod=self.signature_method
-            )
+            hasher = hmac.new(self.signature_secret.encode(), digestmod=self.signature_method)
         else:
             hasher = hashlib.md5()
 
@@ -186,13 +184,9 @@ def get(self, host, request_uri, params=None, auth_type=None):
         if auth_type == 'jwt':
             self._request_headers = self._add_jwt_to_request_headers()
         elif auth_type == 'params':
-            params = dict(
-                params or {}, api_key=self.api_key, api_secret=self.api_secret
-            )
+            params = dict(params or {}, api_key=self.api_key, api_secret=self.api_secret)
         elif auth_type == 'header':
-            hash = base64.b64encode(
-                f"{self.api_key}:{self.api_secret}".encode("utf-8")
-            ).decode("ascii")
+            hash = base64.b64encode(f"{self.api_key}:{self.api_secret}".encode("utf-8")).decode("ascii")
             self._request_headers = dict(self.headers or {}, Authorization=f"Basic {hash}")
         else:
             raise InvalidAuthenticationTypeError(
@@ -201,47 +195,45 @@ def get(self, host, request_uri, params=None, auth_type=None):
 
         logger.debug(f"GET to {repr(uri)} with params {repr(params)}, headers {repr(self._request_headers)}")
         return self.parse(
-            host, 
-            self.session.get(uri, params=params, headers=self._request_headers, timeout=self.timeout))
+            host, self.session.get(uri, params=params, headers=self._request_headers, timeout=self.timeout)
+        )
 
     def post(self, host, request_uri, params, auth_type=None, body_is_json=True, supports_signature_auth=False):
         """
         Low-level method to make a post request to an API server.
         This method automatically adds authentication, picking the first applicable authentication method from the following:
-        - If the supports_signature_auth param is True, and the client was instantiated with a signature_secret, 
+        - If the supports_signature_auth param is True, and the client was instantiated with a signature_secret,
             then signature authentication will be used.
-        :param bool supports_signature_auth: Preferentially use signature authentication if a signature_secret was provided 
+        :param bool supports_signature_auth: Preferentially use signature authentication if a signature_secret was provided
             when initializing this client.
         """
         uri = f"https://{host}{request_uri}"
         self._request_headers = self.headers
-        
+
         if supports_signature_auth and self.signature_secret:
             params["api_key"] = self.api_key
             params["sig"] = self.signature(params)
         elif auth_type == 'jwt':
             self._request_headers = self._add_jwt_to_request_headers()
         elif auth_type == 'params':
-            params = dict(
-                params, api_key=self.api_key, api_secret=self.api_secret
-            )
+            params = dict(params, api_key=self.api_key, api_secret=self.api_secret)
         elif auth_type == 'header':
-            hash = base64.b64encode(
-                f"{self.api_key}:{self.api_secret}".encode("utf-8")
-            ).decode("ascii")
+            hash = base64.b64encode(f"{self.api_key}:{self.api_secret}".encode("utf-8")).decode("ascii")
             self._request_headers = dict(self.headers or {}, Authorization=f"Basic {hash}")
         else:
             raise InvalidAuthenticationTypeError(
                 f'Invalid authentication type. Must be one of "jwt", "header" or "params".'
             )
-        
+
         logger.debug(f"POST to {repr(uri)} with params {repr(params)}, headers {repr(self._request_headers)}")
         if body_is_json:
             return self.parse(
-                host, self.session.post(uri, json=params, headers=self._request_headers, timeout=self.timeout))
+                host, self.session.post(uri, json=params, headers=self._request_headers, timeout=self.timeout)
+            )
         else:
             return self.parse(
-                host, self.session.post(uri, data=params, headers=self._request_headers, timeout=self.timeout))
+                host, self.session.post(uri, data=params, headers=self._request_headers, timeout=self.timeout)
+            )
 
     def put(self, host, request_uri, params, auth_type=None):
         uri = f"https://{host}{request_uri}"
@@ -250,9 +242,7 @@ def put(self, host, request_uri, params, auth_type=None):
         if auth_type == 'jwt':
             self._request_headers = self._add_jwt_to_request_headers()
         elif auth_type == 'header':
-            hash = base64.b64encode(
-                f"{self.api_key}:{self.api_secret}".encode("utf-8")
-            ).decode("ascii")
+            hash = base64.b64encode(f"{self.api_key}:{self.api_secret}".encode("utf-8")).decode("ascii")
             self._request_headers = dict(self._request_headers or {}, Authorization=f"Basic {hash}")
         else:
             raise InvalidAuthenticationTypeError(
@@ -269,27 +259,21 @@ def delete(self, host, request_uri, auth_type=None):
 
         if auth_type == 'jwt':
             self._request_headers = self._add_jwt_to_request_headers()
-        elif auth_type =='header':
-            hash = base64.b64encode(
-                f"{self.api_key}:{self.api_secret}".encode("utf-8")
-            ).decode("ascii")
+        elif auth_type == 'header':
+            hash = base64.b64encode(f"{self.api_key}:{self.api_secret}".encode("utf-8")).decode("ascii")
             self._request_headers = dict(self._request_headers or {}, Authorization=f"Basic {hash}")
         else:
             raise InvalidAuthenticationTypeError(
                 f'Invalid authentication type. Must be one of "jwt", "header" or "params".'
             )
 
         logger.debug(f"DELETE to {repr(uri)} with headers {repr(self._request_headers)}")
-        return self.parse(
-            host, self.session.delete(uri, headers=self._request_headers, timeout=self.timeout)
-        )
+        return self.parse(host, self.session.delete(uri, headers=self._request_headers, timeout=self.timeout))
 
     def parse(self, host, response):
         logger.debug(f"Response headers {repr(response.headers)}")
         if response.status_code == 401:
-            raise AuthenticationError(
-                "Authentication failed. Check you're using a valid authentication method."
-            )
+            raise AuthenticationError("Authentication failed. Check you're using a valid authentication method.")
         elif response.status_code == 204:
             return None
         elif 200 <= response.status_code < 300:
@@ -301,22 +285,16 @@ def parse(self, host, response):
             else:
                 return response.content
         elif 400 <= response.status_code < 500:
-            logger.warning(
-                f"Client error: {response.status_code} {repr(response.content)}"
-            )
+            logger.warning(f"Client error: {response.status_code} {repr(response.content)}")
             message = f"{response.status_code} response from {host}"
 
             # Test for standard error format:
             try:
                 error_data = response.json()
-                if (
-                    "type" in error_data
-                    and "title" in error_data
-                    and "detail" in error_data
-                ):
-                    title=error_data["title"]
-                    detail=error_data["detail"]
-                    type=error_data["type"]
+                if "type" in error_data and "title" in error_data and "detail" in error_data:
+                    title = error_data["title"]
+                    detail = error_data["detail"]
+                    type = error_data["type"]
                     message = f"{title}: {detail} ({type})"
 
             except JSONDecodeError:
@@ -342,7 +320,7 @@ def _generate_application_jwt(self):
         token = jwt.encode(payload, self._private_key, algorithm="RS256")
 
         # If token is string transform it to byte type
-        if(type(token) is str):
+        if type(token) is str:
             token = bytes(token, 'utf-8')
 
         return token

src/vonage/ncco_builder/connect_endpoints.py

@@ -0,0 +1,50 @@
+from pydantic import BaseModel, HttpUrl, AnyUrl, Field, constr
+from typing import Optional, Dict
+from typing_extensions import Literal
+
+
+class ConnectEndpoints:
+    class Endpoint(BaseModel):
+        type: str = 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]]
+
+    class AppEndpoint(Endpoint):
+        type = Field('app', const=True)
+        user: str
+
+    class WebsocketEndpoint(Endpoint):
+        type = Field('websocket', const=True)
+        uri: AnyUrl
+        contentType: Literal['audio/l16;rate=16000', 'audio/l16;rate=8000']
+        headers: Optional[dict]
+
+    class SipEndpoint(Endpoint):
+        type = Field('sip', const=True)
+        uri: str
+        headers: Optional[dict]
+
+    class VbcEndpoint(Endpoint):
+        type = Field('vbc', const=True)
+        extension: str
+
+    @classmethod
+    def create_endpoint_model_from_dict(cls, d) -> Endpoint:
+        if d['type'] == 'phone':
+            return cls.PhoneEndpoint.parse_obj(d)
+        elif d['type'] == 'app':
+            return cls.AppEndpoint.parse_obj(d)
+        elif d['type'] == 'websocket':
+            return cls.WebsocketEndpoint.parse_obj(d)
+        elif d['type'] == 'sip':
+            return cls.WebsocketEndpoint.parse_obj(d)
+        elif d['type'] == 'vbc':
+            return cls.WebsocketEndpoint.parse_obj(d)
+        else:
+            raise ValueError(
+                'Invalid "type" specified for endpoint object. Cannot create a ConnectEndpoints.Endpoint model.'
+            )

src/vonage/ncco_builder/input_types.py

@@ -0,0 +1,26 @@
+from pydantic import BaseModel, confloat, conint
+from typing import Optional, List
+
+
+class InputTypes:
+    class Dtmf(BaseModel):
+        timeOut: Optional[conint(ge=0, le=10)]
+        maxDigits: Optional[conint(ge=1, le=20)]
+        submitOnHash: Optional[bool]
+
+    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]
+
+    @classmethod
+    def create_dtmf_model(cls, dict) -> Dtmf:
+        return cls.Dtmf.parse_obj(dict)
+
+    @classmethod
+    def create_speech_model(cls, dict) -> Speech:
+        return cls.Speech.parse_obj(dict)