diff --git a/tests/unit/base/test_version.py b/tests/unit/base/test_version.py index 4484cbdd5..50e869cb7 100644 --- a/tests/unit/base/test_version.py +++ b/tests/unit/base/test_version.py @@ -119,3 +119,142 @@ def test_delete_not_found(self): ) self.assertIn("Unable to delete record", str(context.exception)) + + def test_fetch_with_response_info(self): + self.holodeck.mock( + Response(200, '{"sid": "AC123", "name": "Test Account"}', {"X-Custom-Header": "test-value"}), + Request(url="https://api.twilio.com/2010-04-01/Accounts/AC123.json"), + ) + payload, status_code, headers = self.client.api.v2010.fetch_with_response_info( + method="GET", uri="/Accounts/AC123.json" + ) + + self.assertEqual(payload["sid"], "AC123") + self.assertEqual(payload["name"], "Test Account") + self.assertEqual(status_code, 200) + self.assertIn("X-Custom-Header", headers) + self.assertEqual(headers["X-Custom-Header"], "test-value") + + def test_update_with_response_info(self): + self.holodeck.mock( + Response(200, '{"sid": "AC123", "name": "Updated Account"}', {"X-Update-Header": "updated"}), + Request( + method="POST", + url="https://api.twilio.com/2010-04-01/Accounts/AC123.json", + ), + ) + payload, status_code, headers = self.client.api.v2010.update_with_response_info( + method="POST", uri="/Accounts/AC123.json", data={"name": "Updated Account"} + ) + + self.assertEqual(payload["sid"], "AC123") + self.assertEqual(payload["name"], "Updated Account") + self.assertEqual(status_code, 200) + self.assertIn("X-Update-Header", headers) + + def test_delete_with_response_info(self): + self.holodeck.mock( + Response(204, "", {"X-Delete-Header": "deleted"}), + Request( + method="DELETE", + url="https://api.twilio.com/2010-04-01/Accounts/AC123/Messages/MM123.json", + ), + ) + success, status_code, headers = self.client.api.v2010.delete_with_response_info( + method="DELETE", uri="/Accounts/AC123/Messages/MM123.json" + ) + + self.assertTrue(success) + self.assertEqual(status_code, 204) + self.assertIn("X-Delete-Header", headers) + + def test_create_with_response_info(self): + self.holodeck.mock( + Response(201, '{"sid": "MM123", "body": "Hello World"}', {"X-Create-Header": "created"}), + Request( + method="POST", + url="https://api.twilio.com/2010-04-01/Accounts/AC123/Messages.json", + ), + ) + payload, status_code, headers = self.client.api.v2010.create_with_response_info( + method="POST", uri="/Accounts/AC123/Messages.json", data={"body": "Hello World"} + ) + + self.assertEqual(payload["sid"], "MM123") + self.assertEqual(payload["body"], "Hello World") + self.assertEqual(status_code, 201) + self.assertIn("X-Create-Header", headers) + + def test_page_with_response_info(self): + self.holodeck.mock( + Response(200, '{"messages": [], "next_page_uri": null}', {"X-Page-Header": "page"}), + Request(url="https://api.twilio.com/2010-04-01/Accounts/AC123/Messages.json"), + ) + response, status_code, headers = self.client.api.v2010.page_with_response_info( + method="GET", uri="/Accounts/AC123/Messages.json" + ) + + self.assertIsNotNone(response) + self.assertEqual(status_code, 200) + self.assertIn("X-Page-Header", headers) + + def test_fetch_with_response_info_error(self): + self.holodeck.mock( + Response(404, '{"message": "Resource not found"}'), + Request(url="https://api.twilio.com/2010-04-01/Accounts/AC456.json"), + ) + + with self.assertRaises(Exception) as context: + self.client.api.v2010.fetch_with_response_info( + method="GET", uri="/Accounts/AC456.json" + ) + + self.assertIn("Unable to fetch record", str(context.exception)) + + def test_update_with_response_info_error(self): + self.holodeck.mock( + Response(400, '{"message": "Invalid request"}'), + Request( + method="POST", + url="https://api.twilio.com/2010-04-01/Accounts/AC123.json", + ), + ) + + with self.assertRaises(Exception) as context: + self.client.api.v2010.update_with_response_info( + method="POST", uri="/Accounts/AC123.json", data={"invalid": "data"} + ) + + self.assertIn("Unable to update record", str(context.exception)) + + def test_delete_with_response_info_error(self): + self.holodeck.mock( + Response(404, '{"message": "Resource not found"}'), + Request( + method="DELETE", + url="https://api.twilio.com/2010-04-01/Accounts/AC123/Messages/MM456.json", + ), + ) + + with self.assertRaises(Exception) as context: + self.client.api.v2010.delete_with_response_info( + method="DELETE", uri="/Accounts/AC123/Messages/MM456.json" + ) + + self.assertIn("Unable to delete record", str(context.exception)) + + def test_create_with_response_info_error(self): + self.holodeck.mock( + Response(400, '{"message": "Invalid request"}'), + Request( + method="POST", + url="https://api.twilio.com/2010-04-01/Accounts/AC123/Messages.json", + ), + ) + + with self.assertRaises(Exception) as context: + self.client.api.v2010.create_with_response_info( + method="POST", uri="/Accounts/AC123/Messages.json", data={"invalid": "data"} + ) + + self.assertIn("Unable to create record", str(context.exception)) diff --git a/tests/unit/http/test_api_response.py b/tests/unit/http/test_api_response.py new file mode 100644 index 000000000..099f3a7ac --- /dev/null +++ b/tests/unit/http/test_api_response.py @@ -0,0 +1,98 @@ +import unittest +from twilio.base.api_response import ApiResponse + + +class TestApiResponse(unittest.TestCase): + def test_initialization(self): + """Test ApiResponse initialization""" + data = {'sid': 'AC123', 'friendly_name': 'Test'} + response = ApiResponse( + data=data, + status_code=200, + headers={'Content-Type': 'application/json'} + ) + + self.assertEqual(response.data, data) + self.assertEqual(response.status_code, 200) + self.assertEqual(response.headers['Content-Type'], 'application/json') + + def test_repr(self): + """Test string representation""" + response = ApiResponse(data={'test': 'data'}, status_code=201, headers={}) + repr_str = repr(response) + self.assertIn('201', repr_str) + self.assertIn('dict', repr_str) + + def test_str(self): + """Test human-readable string representation""" + response = ApiResponse(data={'test': 'data'}, status_code=200, headers={}) + str_repr = str(response) + self.assertIn('200', str_repr) + self.assertIn('dict', str_repr) + + def test_with_list_data(self): + """Test ApiResponse with list data""" + data = [{'sid': 'AC1'}, {'sid': 'AC2'}] + response = ApiResponse(data=data, status_code=200, headers={}) + self.assertEqual(len(response.data), 2) + self.assertEqual(response.status_code, 200) + + def test_with_boolean_data(self): + """Test ApiResponse with boolean data (delete operations)""" + response = ApiResponse(data=True, status_code=204, headers={}) + self.assertTrue(response.data) + self.assertEqual(response.status_code, 204) + + def test_with_different_status_codes(self): + """Test ApiResponse with various status codes""" + # Test 201 Created + response_201 = ApiResponse(data={'created': True}, status_code=201, headers={}) + self.assertEqual(response_201.status_code, 201) + + # Test 204 No Content + response_204 = ApiResponse(data=True, status_code=204, headers={}) + self.assertEqual(response_204.status_code, 204) + + # Test 200 OK + response_200 = ApiResponse(data={'ok': True}, status_code=200, headers={}) + self.assertEqual(response_200.status_code, 200) + + def test_headers_access(self): + """Test accessing various headers""" + headers = { + 'Content-Type': 'application/json', + 'X-RateLimit-Remaining': '100', + 'X-RateLimit-Limit': '1000' + } + response = ApiResponse(data={'test': 'data'}, status_code=200, headers=headers) + + self.assertEqual(response.headers['Content-Type'], 'application/json') + self.assertEqual(response.headers['X-RateLimit-Remaining'], '100') + self.assertEqual(response.headers['X-RateLimit-Limit'], '1000') + + def test_empty_headers(self): + """Test ApiResponse with empty headers""" + response = ApiResponse(data={'test': 'data'}, status_code=200, headers={}) + self.assertEqual(response.headers, {}) + + def test_data_attribute_types(self): + """Test that data attribute can hold various types""" + # Dictionary data + dict_response = ApiResponse(data={'key': 'value'}, status_code=200, headers={}) + self.assertIsInstance(dict_response.data, dict) + + # List data + list_response = ApiResponse(data=[1, 2, 3], status_code=200, headers={}) + self.assertIsInstance(list_response.data, list) + + # Boolean data + bool_response = ApiResponse(data=True, status_code=204, headers={}) + self.assertIsInstance(bool_response.data, bool) + + # None data + none_response = ApiResponse(data=None, status_code=204, headers={}) + self.assertIsNone(none_response.data) + + +if __name__ == '__main__': + unittest.main() diff --git a/twilio/base/api_response.py b/twilio/base/api_response.py new file mode 100644 index 000000000..216f2457d --- /dev/null +++ b/twilio/base/api_response.py @@ -0,0 +1,50 @@ +""" +ApiResponse class for wrapping API responses with metadata +""" +from typing import Any, Dict, Generic, TypeVar + +T = TypeVar('T') + + +class ApiResponse(Generic[T]): + """ + Wrapper for API responses that includes HTTP metadata. + + This class is returned by *_with_http_info methods and provides access to: + - The response data (resource instance, list, or boolean) + - HTTP status code + - Response headers + + Attributes: + data: The response data (instance, list, or boolean for delete operations) + status_code: HTTP status code from the response + headers: Dictionary of response headers + + Example: + >>> response = client.accounts.create_with_http_info(friendly_name="Test") + >>> print(response.status_code) # 201 + >>> print(response.headers['Content-Type']) # application/json + >>> account = response.data + >>> print(account.sid) + """ + + def __init__(self, data: T, status_code: int, headers: Dict[str, str]): + """ + Initialize an ApiResponse + + Args: + data: The response payload (instance, list, or boolean) + status_code: HTTP status code (e.g., 200, 201, 204) + headers: Dictionary of response headers + """ + self.data = data + self.status_code = status_code + self.headers = headers + + def __repr__(self) -> str: + """String representation of the ApiResponse""" + return f"ApiResponse(status_code={self.status_code}, data={type(self.data).__name__})" + + def __str__(self) -> str: + """Human-readable string representation""" + return f"" diff --git a/twilio/base/version.py b/twilio/base/version.py index aead1e020..b1da2fd39 100644 --- a/twilio/base/version.py +++ b/twilio/base/version.py @@ -166,6 +166,72 @@ async def fetch_async( ) return self._parse_fetch(method, uri, response) + def fetch_with_response_info( + self, + method: str, + uri: str, + params: Optional[Dict[str, object]] = None, + data: Optional[Dict[str, object]] = None, + headers: Optional[Dict[str, str]] = None, + auth: Optional[Tuple[str, str]] = None, + timeout: Optional[float] = None, + allow_redirects: bool = False, + ) -> Tuple[Any, int, Dict[str, str]]: + """ + Fetch a resource and return response metadata + + Returns: + tuple: (payload_dict, status_code, headers_dict) + - payload_dict: The JSON response body as a dictionary + - status_code: HTTP status code (typically 200) + - headers_dict: Response headers as a dictionary + """ + response = self.request( + method, + uri, + params=params, + data=data, + headers=headers, + auth=auth, + timeout=timeout, + allow_redirects=allow_redirects, + ) + payload = self._parse_fetch(method, uri, response) + return payload, response.status_code, dict(response.headers) + + async def fetch_with_response_info_async( + self, + method: str, + uri: str, + params: Optional[Dict[str, object]] = None, + data: Optional[Dict[str, object]] = None, + headers: Optional[Dict[str, str]] = None, + auth: Optional[Tuple[str, str]] = None, + timeout: Optional[float] = None, + allow_redirects: bool = False, + ) -> Tuple[Any, int, Dict[str, str]]: + """ + Asynchronously fetch a resource and return response metadata + + Returns: + tuple: (payload_dict, status_code, headers_dict) + - payload_dict: The JSON response body as a dictionary + - status_code: HTTP status code (typically 200) + - headers_dict: Response headers as a dictionary + """ + response = await self.request_async( + method, + uri, + params=params, + data=data, + headers=headers, + auth=auth, + timeout=timeout, + allow_redirects=allow_redirects, + ) + payload = self._parse_fetch(method, uri, response) + return (payload, response.status_code, dict(response.headers)) + def _parse_update(self, method: str, uri: str, response: Response) -> Any: """ Parses update response JSON @@ -229,6 +295,72 @@ async def update_async( return self._parse_update(method, uri, response) + def update_with_response_info( + self, + method: str, + uri: str, + params: Optional[Dict[str, object]] = None, + data: Optional[Dict[str, object]] = None, + headers: Optional[Dict[str, str]] = None, + auth: Optional[Tuple[str, str]] = None, + timeout: Optional[float] = None, + allow_redirects: bool = False, + ) -> Tuple[Any, int, Dict[str, str]]: + """ + Update a resource and return response metadata + + Returns: + tuple: (payload_dict, status_code, headers_dict) + - payload_dict: The JSON response body as a dictionary + - status_code: HTTP status code (typically 200) + - headers_dict: Response headers as a dictionary + """ + response = self.request( + method, + uri, + params=params, + data=data, + headers=headers, + auth=auth, + timeout=timeout, + allow_redirects=allow_redirects, + ) + payload = self._parse_update(method, uri, response) + return (payload, response.status_code, dict(response.headers)) + + async def update_with_response_info_async( + self, + method: str, + uri: str, + params: Optional[Dict[str, object]] = None, + data: Optional[Dict[str, object]] = None, + headers: Optional[Dict[str, str]] = None, + auth: Optional[Tuple[str, str]] = None, + timeout: Optional[float] = None, + allow_redirects: bool = False, + ) -> Tuple[Any, int, Dict[str, str]]: + """ + Asynchronously update a resource and return response metadata + + Returns: + tuple: (payload_dict, status_code, headers_dict) + - payload_dict: The JSON response body as a dictionary + - status_code: HTTP status code (typically 200) + - headers_dict: Response headers as a dictionary + """ + response = await self.request_async( + method, + uri, + params=params, + data=data, + headers=headers, + auth=auth, + timeout=timeout, + allow_redirects=allow_redirects, + ) + payload = self._parse_update(method, uri, response) + return (payload, response.status_code, dict(response.headers)) + def _parse_delete(self, method: str, uri: str, response: Response) -> bool: """ Parses delete response JSON @@ -292,6 +424,72 @@ async def delete_async( return self._parse_delete(method, uri, response) + def delete_with_response_info( + self, + method: str, + uri: str, + params: Optional[Dict[str, object]] = None, + data: Optional[Dict[str, object]] = None, + headers: Optional[Dict[str, str]] = None, + auth: Optional[Tuple[str, str]] = None, + timeout: Optional[float] = None, + allow_redirects: bool = False, + ) -> Tuple[bool, int, Dict[str, str]]: + """ + Delete a resource and return response metadata + + Returns: + tuple: (success_boolean, status_code, headers_dict) + - success_boolean: True if deletion was successful (2XX response) + - status_code: HTTP status code (typically 204 for successful delete) + - headers_dict: Response headers as a dictionary + """ + response = self.request( + method, + uri, + params=params, + data=data, + headers=headers, + auth=auth, + timeout=timeout, + allow_redirects=allow_redirects, + ) + success = self._parse_delete(method, uri, response) + return (success, response.status_code, dict(response.headers)) + + async def delete_with_response_info_async( + self, + method: str, + uri: str, + params: Optional[Dict[str, object]] = None, + data: Optional[Dict[str, object]] = None, + headers: Optional[Dict[str, str]] = None, + auth: Optional[Tuple[str, str]] = None, + timeout: Optional[float] = None, + allow_redirects: bool = False, + ) -> Tuple[bool, int, Dict[str, str]]: + """ + Asynchronously delete a resource and return response metadata + + Returns: + tuple: (success_boolean, status_code, headers_dict) + - success_boolean: True if deletion was successful (2XX response) + - status_code: HTTP status code (typically 204 for successful delete) + - headers_dict: Response headers as a dictionary + """ + response = await self.request_async( + method, + uri, + params=params, + data=data, + headers=headers, + auth=auth, + timeout=timeout, + allow_redirects=allow_redirects, + ) + success = self._parse_delete(method, uri, response) + return (success, response.status_code, dict(response.headers)) + def read_limits( self, limit: Optional[int] = None, page_size: Optional[int] = None ) -> Dict[str, object]: @@ -361,6 +559,70 @@ async def page_async( allow_redirects=allow_redirects, ) + def page_with_response_info( + self, + method: str, + uri: str, + params: Optional[Dict[str, object]] = None, + data: Optional[Dict[str, object]] = None, + headers: Optional[Dict[str, str]] = None, + auth: Optional[Tuple[str, str]] = None, + timeout: Optional[float] = None, + allow_redirects: bool = False, + ) -> Tuple[Response, int, Dict[str, str]]: + """ + Fetch a page and return response metadata + + Returns: + tuple: (response_object, status_code, headers_dict) + - response_object: The Response object (not parsed JSON) + - status_code: HTTP status code + - headers_dict: Response headers as a dictionary + """ + response = self.request( + method, + uri, + params=params, + data=data, + headers=headers, + auth=auth, + timeout=timeout, + allow_redirects=allow_redirects, + ) + return (response, response.status_code, dict(response.headers)) + + async def page_with_response_info_async( + self, + method: str, + uri: str, + params: Optional[Dict[str, object]] = None, + data: Optional[Dict[str, object]] = None, + headers: Optional[Dict[str, str]] = None, + auth: Optional[Tuple[str, str]] = None, + timeout: Optional[float] = None, + allow_redirects: bool = False, + ) -> Tuple[Response, int, Dict[str, str]]: + """ + Asynchronously fetch a page and return response metadata + + Returns: + tuple: (response_object, status_code, headers_dict) + - response_object: The Response object (not parsed JSON) + - status_code: HTTP status code + - headers_dict: Response headers as a dictionary + """ + response = await self.request_async( + method, + uri, + params=params, + data=data, + headers=headers, + auth=auth, + timeout=timeout, + allow_redirects=allow_redirects, + ) + return (response, response.status_code, dict(response.headers)) + def stream( self, page: Optional[Page], @@ -487,3 +749,69 @@ async def create_async( allow_redirects=allow_redirects, ) return self._parse_create(method, uri, response) + + def create_with_response_info( + self, + method: str, + uri: str, + params: Optional[Dict[str, object]] = None, + data: Optional[Dict[str, object]] = None, + headers: Optional[Dict[str, str]] = None, + auth: Optional[Tuple[str, str]] = None, + timeout: Optional[float] = None, + allow_redirects: bool = False, + ) -> Tuple[Any, int, Dict[str, str]]: + """ + Create a resource and return response metadata + + Returns: + tuple: (payload_dict, status_code, headers_dict) + - payload_dict: The JSON response body as a dictionary + - status_code: HTTP status code (e.g., 201) + - headers_dict: Response headers as a dictionary + """ + response = self.request( + method, + uri, + params=params, + data=data, + headers=headers, + auth=auth, + timeout=timeout, + allow_redirects=allow_redirects, + ) + payload = self._parse_create(method, uri, response) + return (payload, response.status_code, dict(response.headers)) + + async def create_with_response_info_async( + self, + method: str, + uri: str, + params: Optional[Dict[str, object]] = None, + data: Optional[Dict[str, object]] = None, + headers: Optional[Dict[str, str]] = None, + auth: Optional[Tuple[str, str]] = None, + timeout: Optional[float] = None, + allow_redirects: bool = False, + ) -> Tuple[Any, int, Dict[str, str]]: + """ + Asynchronously create a resource and return response metadata + + Returns: + tuple: (payload_dict, status_code, headers_dict) + - payload_dict: The JSON response body as a dictionary + - status_code: HTTP status code (e.g., 201) + - headers_dict: Response headers as a dictionary + """ + response = await self.request_async( + method, + uri, + params=params, + data=data, + headers=headers, + auth=auth, + timeout=timeout, + allow_redirects=allow_redirects, + ) + payload = self._parse_create(method, uri, response) + return (payload, response.status_code, dict(response.headers))