refactoring all rest call logic into a single method, send_request

CHANGES.md

@@ -8,6 +8,7 @@ Breaking changes:
 - Removed `Pay` action from NCCO builder
 - Removed `Redact` class and support for the Redact API as it's a dev preview product that's unsupported in the SDK
 - Removed `ApplicationV2` class as V1 has been end-of-life for a significant amount of time and this new naming is in line with other APIs. Please use `Application` instead
+- Removed `Account.get_sms_pricing` and `Account.get_voice_pricing` methods as the endpoints they call been deprecated for multiple years.
 
 # 3.9.0
 - Dropped support for Python 3.7 as it's end-of-life and no longer receiving security updates

src/vonage/account.py

@@ -6,14 +6,20 @@ class Account:
     pricing_auth_type = 'params'
     secrets_auth_type = 'header'
 
+    account_sent_data_type = 'data'
+    pricing_sent_data_type = 'query'
+    secrets_sent_data_type = 'json'
+
     allowed_pricing_types = {'sms', 'sms-transit', 'voice'}
 
     def __init__(self, client):
         self._client = client
 
     def get_balance(self):
         return self._client.get(
-            self._client.host(), "/account/get-balance", auth_type=Account.account_auth_type
+            self._client.host(),
+            "/account/get-balance",
+            auth_type=Account.account_auth_type,
         )
 
     def topup(self, params=None, **kwargs):
@@ -22,7 +28,7 @@ def topup(self, params=None, **kwargs):
             "/account/top-up",
             params or kwargs,
             auth_type=Account.account_auth_type,
-            body_is_json=False,
+            sent_data_type=Account.account_sent_data_type,
         )
 
     def get_country_pricing(self, country_code: str, type: str = 'sms'):
@@ -32,6 +38,7 @@ def get_country_pricing(self, country_code: str, type: str = 'sms'):
             f"/account/get-pricing/outbound/{type}",
             {"country": country_code},
             auth_type=Account.pricing_auth_type,
+            sent_data_type=Account.pricing_sent_data_type,
         )
 
     def get_all_countries_pricing(self, type: str = 'sms'):
@@ -49,6 +56,7 @@ def get_prefix_pricing(self, prefix: str, type: str = 'sms'):
             f"/account/get-prefix-pricing/outbound/{type}",
             {"prefix": prefix},
             auth_type=Account.pricing_auth_type,
+            sent_data_type=Account.pricing_sent_data_type,
         )
 
     def update_default_sms_webhook(self, params=None, **kwargs):
@@ -57,7 +65,7 @@ def update_default_sms_webhook(self, params=None, **kwargs):
             "/account/settings",
             params or kwargs,
             auth_type=Account.account_auth_type,
-            body_is_json=False,
+            sent_data_type=Account.account_sent_data_type,
         )
 
     def list_secrets(self, api_key):
@@ -81,7 +89,6 @@ def create_secret(self, api_key, secret):
             f"/accounts/{api_key}/secrets",
             body,
             auth_type=Account.secrets_auth_type,
-            body_is_json=False,
         )
 
     def revoke_secret(self, api_key, secret_id):

src/vonage/client.py

@@ -25,7 +25,6 @@
 import base64
 import hashlib
 import hmac
-import os
 import time
 
 from requests import Response
@@ -217,55 +216,55 @@ def signature(self, params):
 
         return hasher.hexdigest()
 
-    def get(self, host, request_uri, params=None, auth_type=None):
-        uri = f"https://{host}{request_uri}"
-        self._request_headers = self.headers
-
-        if auth_type == 'jwt':
-            self._request_headers['Authorization'] = self._create_jwt_auth_string()
-        elif auth_type == 'params':
-            params = dict(
-                params or {},
-                api_key=self.api_key,
-                api_secret=self.api_secret,
-            )
-        elif auth_type == 'header':
-            self._request_headers['Authorization'] = self._create_header_auth_string()
-        else:
-            raise InvalidAuthenticationTypeError(
-                f'Invalid authentication type. Must be one of "jwt", "header" or "params".'
-            )
-
-        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,
-            ),
-        )
+    def get(self, host, request_uri, params={}, auth_type=None, sent_data_type='json'):
+        return self.send_request('GET', host, request_uri, params, auth_type, sent_data_type)
 
     def post(
         self,
         host,
         request_uri,
-        params,
+        params=None,
         auth_type=None,
-        body_is_json=True,
+        sent_data_type='json',
+        supports_signature_auth=False,
+    ):
+        return self.send_request(
+            'POST', host, request_uri, params, auth_type, sent_data_type, supports_signature_auth
+        )
+
+    def put(self, host, request_uri, params, auth_type=None):
+        return self.send_request('PUT', host, request_uri, params, auth_type)
+
+    def patch(self, host, request_uri, params, auth_type=None):
+        return self.send_request('PATCH', host, request_uri, params, auth_type)
+
+    def delete(self, host, request_uri, params=None, auth_type=None):
+        return self.send_request('DELETE', host, request_uri, params, auth_type)
+
+    def send_request(
+        self,
+        request_type: str,
+        host: str,
+        request_uri: str,
+        params: dict = {},
+        auth_type=None,
+        sent_data_type='json',
         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,
-            then signature authentication will be used.
-        :param bool supports_signature_auth: Preferentially use signature authentication if a signature_secret was provided
-            when initializing this client.
+        Low-level method to make a request to an API server.
+        The supports_signature_auth parameter lets you preferentially use signature authentication if a
+        signature_secret was provided when initializing this client (only for the SMS API).
         """
+
+        allowed_request_types = {'GET', 'POST', 'PUT', 'PATCH', 'DELETE'}
+        if request_type not in allowed_request_types:
+            raise ClientError('Invalid request type.')
+
+        allowed_sent_data_types = {'json', 'data', 'query'}
+        if sent_data_type not in allowed_sent_data_types:
+            raise ClientError('Invalid sent_data type.')
+
         uri = f"https://{host}{request_uri}"
         self._request_headers = self.headers
 
@@ -288,106 +287,42 @@ def post(
             )
 
         logger.debug(
-            f"POST to {repr(uri)} with params {repr(params)}, headers {repr(self._request_headers)}"
+            f'{request_type} to {repr(uri)} with params {repr(params)}, headers {repr(self._request_headers)}'
         )
-        if body_is_json:
+        if sent_data_type == 'json':
             return self.parse(
                 host,
-                self.session.post(
+                self.session.request(
+                    request_type,
                     uri,
                     json=params,
                     headers=self._request_headers,
                     timeout=self.timeout,
                 ),
             )
-        else:
+        elif sent_data_type == 'data':
             return self.parse(
                 host,
-                self.session.post(
+                self.session.request(
+                    request_type,
                     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}"
-        self._request_headers = self.headers
-
-        if auth_type == 'jwt':
-            self._request_headers['Authorization'] = self._create_jwt_auth_string()
-        elif auth_type == 'header':
-            self._request_headers['Authorization'] = self._create_header_auth_string()
-        else:
-            raise InvalidAuthenticationTypeError(
-                f'Invalid authentication type. Must be one of "jwt" or "header".'
-            )
-
-        logger.debug(
-            f"PUT to {repr(uri)} with params {repr(params)}, headers {repr(self._request_headers)}"
-        )
-        # All APIs that currently use put methods require a json-formatted body so don't need to check this
-        return self.parse(
-            host,
-            self.session.put(
-                uri,
-                json=params,
-                headers=self._request_headers,
-                timeout=self.timeout,
-            ),
-        )
-
-    def patch(self, host, request_uri, params, auth_type=None):
-        uri = f"https://{host}{request_uri}"
-        self._request_headers = self.headers
-
-        if auth_type == 'jwt':
-            self._request_headers['Authorization'] = self._create_jwt_auth_string()
-        elif auth_type == 'header':
-            self._request_headers['Authorization'] = self._create_header_auth_string()
-        else:
-            raise InvalidAuthenticationTypeError(f"""Invalid authentication type.""")
-
-        logger.debug(
-            f"PATCH to {repr(uri)} with params {repr(params)}, headers {repr(self._request_headers)}"
-        )
-        # Only newer APIs (that expect json-bodies) currently use this method, so we will always send a json-formatted body
-        return self.parse(
-            host,
-            self.session.patch(
-                uri,
-                json=params,
-                headers=self._request_headers,
-            ),
-        )
-
-    def delete(self, host, request_uri, params=None, auth_type=None):
-        uri = f"https://{host}{request_uri}"
-        self._request_headers = self.headers
-
-        if auth_type == 'jwt':
-            self._request_headers['Authorization'] = self._create_jwt_auth_string()
-        elif auth_type == 'header':
-            self._request_headers['Authorization'] = self._create_header_auth_string()
         else:
-            raise InvalidAuthenticationTypeError(
-                f'Invalid authentication type. Must be one of "jwt", "header" or "params".'
+            return self.parse(
+                host,
+                self.session.request(
+                    request_type,
+                    uri,
+                    params=params,
+                    headers=self._request_headers,
+                    timeout=self.timeout,
+                ),
             )
 
-        logger.debug(f"DELETE to {repr(uri)} with headers {repr(self._request_headers)}")
-        if params is not None:
-            logger.debug(f"DELETE call has params {repr(params)}")
-        return self.parse(
-            host,
-            self.session.delete(
-                uri,
-                headers=self._request_headers,
-                timeout=self.timeout,
-                params=params,
-            ),
-        )
-
     def parse(self, host, response: Response):
         logger.debug(f"Response headers {repr(response.headers)}")
         if response.status_code == 401:

src/vonage/number_insight.py

@@ -4,6 +4,7 @@
 
 class NumberInsight:
     auth_type = 'params'
+    sent_data_type = 'query'
 
     def __init__(self, client):
         self._client = client
@@ -14,6 +15,7 @@ def get_basic_number_insight(self, params=None, **kwargs):
             "/ni/basic/json",
             params or kwargs,
             auth_type=NumberInsight.auth_type,
+            sent_data_type=NumberInsight.sent_data_type,
         )
         self.check_for_error(response)
 
@@ -25,6 +27,7 @@ def get_standard_number_insight(self, params=None, **kwargs):
             "/ni/standard/json",
             params or kwargs,
             auth_type=NumberInsight.auth_type,
+            sent_data_type=NumberInsight.sent_data_type,
         )
         self.check_for_error(response)
 
@@ -36,6 +39,7 @@ def get_advanced_number_insight(self, params=None, **kwargs):
             "/ni/advanced/json",
             params or kwargs,
             auth_type=NumberInsight.auth_type,
+            sent_data_type=NumberInsight.sent_data_type,
         )
         self.check_for_error(response)
 
@@ -50,6 +54,7 @@ def get_async_advanced_number_insight(self, params=None, **kwargs):
             "/ni/advanced/async/json",
             params or kwargs,
             auth_type=NumberInsight.auth_type,
+            sent_data_type=NumberInsight.sent_data_type,
         )
         print(json.dumps(response, indent=4))
         self.check_for_async_error(response)

src/vonage/number_management.py

@@ -1,13 +1,19 @@
 class Numbers:
     auth_type = 'header'
-    defaults = {'auth_type': auth_type, 'body_is_json': False}
+    sent_data_type_GET = 'query'
+    sent_data_type_POST = 'data'
+    defaults = {'auth_type': auth_type, 'sent_data_type': sent_data_type_POST}
 
     def __init__(self, client):
         self._client = client
 
     def get_account_numbers(self, params=None, **kwargs):
         return self._client.get(
-            self._client.host(), "/account/numbers", params or kwargs, auth_type=Numbers.auth_type
+            self._client.host(),
+            "/account/numbers",
+            params or kwargs,
+            auth_type=Numbers.auth_type,
+            sent_data_type=Numbers.sent_data_type_GET,
         )
 
     def get_available_numbers(self, country_code, params=None, **kwargs):
@@ -16,6 +22,7 @@ def get_available_numbers(self, country_code, params=None, **kwargs):
             "/number/search",
             dict(params or kwargs, country=country_code),
             auth_type=Numbers.auth_type,
+            sent_data_type=Numbers.sent_data_type_GET,
         )
 
     def buy_number(self, params=None, **kwargs):

src/vonage/short_codes.py

@@ -1,6 +1,6 @@
 class ShortCodes:
     auth_type = 'params'
-    defaults = {'auth_type': auth_type, 'body_is_json': False}
+    defaults = {'auth_type': auth_type, 'sent_data_type': 'data'}
 
     def __init__(self, client):
         self._client = client

src/vonage/sms.py

@@ -5,7 +5,7 @@
 
 
 class Sms:
-    defaults = {'auth_type': 'params', 'body_is_json': False}
+    defaults = {'auth_type': 'params', 'sent_data_type': 'data'}
 
     def __init__(self, client):
         self._client = client

src/vonage/ussd.py

@@ -1,5 +1,5 @@
 class Ussd:
-    defaults = {'auth_type': 'params', 'body_is_json': False}
+    defaults = {'auth_type': 'params', 'sent_data_type': 'data'}
 
     def __init__(self, client):
         self._client = client