Skip to content

Add response_headers to APIException #168

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
chrisballinger wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Add response_headers to APIException #168

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 20 additions & 4 deletions appstoreserverlibrary/api_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -574,16 +574,32 @@ class APIError(IntEnum):

@define
class APIException(Exception):
"""
Exception raised when the App Store Server API returns an error response.

Attributes:
http_status_code: The HTTP status code from the response
api_error: The parsed APIError enum value, if recognized
raw_api_error: The raw error code from the API response
error_message: The error message from the API response
response_headers: HTTP response headers from the error response. This is useful for accessing
rate limiting information such as the 'Retry-After' header when receiving a 429 response.
Note: Header key casing depends on the HTTP client library used. The async client (httpx)
normalizes all header keys to lowercase, while the sync client (requests) uses case-insensitive
access. For portability, use lowercase header keys (e.g., 'retry-after' instead of 'Retry-After').
"""
http_status_code: int
api_error: Optional[APIError]
raw_api_error: Optional[int]
error_message: Optional[str]
response_headers: Optional[Dict[str, str]] = None

def __init__(self, http_status_code: int, raw_api_error: Optional[int] = None, error_message: Optional[str] = None):
def __init__(self, http_status_code: int, raw_api_error: Optional[int] = None, error_message: Optional[str] = None, response_headers: Optional[Dict[str, str]] = None):
self.http_status_code = http_status_code
self.raw_api_error = raw_api_error
self.api_error = None
self.error_message = error_message
self.response_headers = response_headers or {}
try:
if raw_api_error is not None:
self.api_error = APIError(raw_api_error)
Expand Down Expand Up @@ -653,14 +669,14 @@ def _parse_response(self, status_code: int, headers: MutableMapping, json_suppli
else:
# Best effort parsing of the response body
if not 'content-type' in headers or headers['content-type'] != 'application/json':
raise APIException(status_code)
raise APIException(status_code, response_headers=dict(headers))
try:
response_body = json_supplier()
raise APIException(status_code, response_body['errorCode'], response_body['errorMessage'])
raise APIException(status_code, response_body['errorCode'], response_body['errorMessage'], response_headers=dict(headers))
except APIException as e:
raise e
except Exception as e:
raise APIException(status_code) from e
raise APIException(status_code, response_headers=dict(headers)) from e


class AppStoreServerAPIClient(BaseAppStoreServerAPIClient):
Expand Down
63 changes: 62 additions & 1 deletion tests/test_api_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -434,6 +434,64 @@ def test_unknown_error(self):

self.assertFalse(True)

def test_api_exception_includes_response_headers(self):
# Test that response headers are captured in APIException
custom_headers = {
'X-Custom-Header': 'test-value',
'X-Another-Header': 'another-value'
}
client = self.get_client_with_body(b'{"errorCode": 4040010, "errorMessage": "Transaction id not found."}',
'POST',
'https://local-testing-base-url/inApps/v1/notifications/test',
{},
None,
404,
response_headers=custom_headers)
try:
client.request_test_notification()
except APIException as e:
self.assertEqual(404, e.http_status_code)
self.assertEqual(4040010, e.raw_api_error)
self.assertEqual(APIError.TRANSACTION_ID_NOT_FOUND, e.api_error)
self.assertEqual("Transaction id not found.", e.error_message)
# Verify response_headers are captured
self.assertIsNotNone(e.response_headers)
self.assertEqual('test-value', e.response_headers['X-Custom-Header'])
self.assertEqual('another-value', e.response_headers['X-Another-Header'])
self.assertEqual('application/json', e.response_headers['Content-Type'])
return

self.assertFalse(True)

def test_rate_limit_with_retry_after_header(self):
# Test that Retry-After header is accessible for rate limiting
retry_after_time = '1699564800000' # Unix time in milliseconds
rate_limit_headers = {
'Retry-After': retry_after_time,
'X-Rate-Limit-Remaining': '0'
}
client = self.get_client_with_body(b'{"errorCode": 4290000, "errorMessage": "Rate limit exceeded."}',
'POST',
'https://local-testing-base-url/inApps/v1/notifications/test',
{},
None,
429,
response_headers=rate_limit_headers)
try:
client.request_test_notification()
except APIException as e:
self.assertEqual(429, e.http_status_code)
self.assertEqual(4290000, e.raw_api_error)
self.assertEqual(APIError.RATE_LIMIT_EXCEEDED, e.api_error)
self.assertEqual("Rate limit exceeded.", e.error_message)
# Verify Retry-After header is accessible
self.assertIsNotNone(e.response_headers)
self.assertEqual(retry_after_time, e.response_headers['Retry-After'])
self.assertEqual('0', e.response_headers['X-Rate-Limit-Remaining'])
return

self.assertFalse(True)

def test_get_transaction_history_with_unknown_environment(self):
client = self.get_client_with_body_from_file('tests/resources/models/transactionHistoryResponseWithMalformedEnvironment.json',
'GET',
Expand Down Expand Up @@ -723,7 +781,7 @@ def test_get_app_transaction_info_transaction_id_not_found(self):
def get_signing_key(self):
return read_data_from_binary_file('tests/resources/certs/testSigningKey.p8')

def get_client_with_body(self, body: str, expected_method: str, expected_url: str, expected_params: Dict[str, Union[str, List[str]]], expected_json: Dict[str, Any], status_code: int = 200, expected_data: bytes = None, expected_content_type: str = None):
def get_client_with_body(self, body: str, expected_method: str, expected_url: str, expected_params: Dict[str, Union[str, List[str]]], expected_json: Dict[str, Any], status_code: int = 200, expected_data: bytes = None, expected_content_type: str = None, response_headers: Dict[str, str] = None):
signing_key = self.get_signing_key()
client = AppStoreServerAPIClient(signing_key, 'keyId', 'issuerId', 'com.example', Environment.LOCAL_TESTING)
def fake_execute_and_validate_inputs(method: bytes, url: str, params: Dict[str, Union[str, List[str]]], headers: Dict[str, str], json: Dict[str, Any], data: bytes):
Expand Down Expand Up @@ -753,6 +811,9 @@ def fake_execute_and_validate_inputs(method: bytes, url: str, params: Dict[str,
response.status_code = status_code
response.raw = BytesIO(body)
response.headers['Content-Type'] = 'application/json'
if response_headers:
for key, value in response_headers.items():
response.headers[key] = value
return response

client._execute_request = fake_execute_and_validate_inputs
Expand Down
67 changes: 65 additions & 2 deletions tests/test_api_client_async.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -440,6 +440,66 @@ async def test_unknown_error(self):

self.assertFalse(True)

async def test_api_exception_includes_response_headers(self):
# Test that response headers are captured in APIException
custom_headers = {
'X-Custom-Header': 'test-value',
'X-Another-Header': 'another-value'
}
client = self.get_client_with_body(b'{"errorCode": 4040010, "errorMessage": "Transaction id not found."}',
'POST',
'https://local-testing-base-url/inApps/v1/notifications/test',
{},
None,
404,
response_headers=custom_headers)
try:
await client.request_test_notification()
except APIException as e:
self.assertEqual(404, e.http_status_code)
self.assertEqual(4040010, e.raw_api_error)
self.assertEqual(APIError.TRANSACTION_ID_NOT_FOUND, e.api_error)
self.assertEqual("Transaction id not found.", e.error_message)
# Verify response_headers are captured
# Note: httpx normalizes all header keys to lowercase
self.assertIsNotNone(e.response_headers)
self.assertEqual('test-value', e.response_headers['x-custom-header'])
self.assertEqual('another-value', e.response_headers['x-another-header'])
self.assertEqual('application/json', e.response_headers['content-type'])
return

self.assertFalse(True)

async def test_rate_limit_with_retry_after_header(self):
# Test that Retry-After header is accessible for rate limiting
retry_after_time = '1699564800000' # Unix time in milliseconds
rate_limit_headers = {
'Retry-After': retry_after_time,
'X-Rate-Limit-Remaining': '0'
}
client = self.get_client_with_body(b'{"errorCode": 4290000, "errorMessage": "Rate limit exceeded."}',
'POST',
'https://local-testing-base-url/inApps/v1/notifications/test',
{},
None,
429,
response_headers=rate_limit_headers)
try:
await client.request_test_notification()
except APIException as e:
self.assertEqual(429, e.http_status_code)
self.assertEqual(4290000, e.raw_api_error)
self.assertEqual(APIError.RATE_LIMIT_EXCEEDED, e.api_error)
self.assertEqual("Rate limit exceeded.", e.error_message)
# Verify Retry-After header is accessible
# Note: httpx normalizes all header keys to lowercase
self.assertIsNotNone(e.response_headers)
self.assertEqual(retry_after_time, e.response_headers['retry-after'])
self.assertEqual('0', e.response_headers['x-rate-limit-remaining'])
return

self.assertFalse(True)

async def test_get_transaction_history_with_unknown_environment(self):
client = self.get_client_with_body_from_file('tests/resources/models/transactionHistoryResponseWithMalformedEnvironment.json',
'GET',
Expand Down Expand Up @@ -728,7 +788,7 @@ async def test_get_app_transaction_info_transaction_id_not_found(self):
def get_signing_key(self):
return read_data_from_binary_file('tests/resources/certs/testSigningKey.p8')

def get_client_with_body(self, body: str, expected_method: str, expected_url: str, expected_params: Dict[str, Union[str, List[str]]], expected_json: Dict[str, Any], status_code: int = 200, expected_data: bytes = None, expected_content_type: str = None):
def get_client_with_body(self, body: str, expected_method: str, expected_url: str, expected_params: Dict[str, Union[str, List[str]]], expected_json: Dict[str, Any], status_code: int = 200, expected_data: bytes = None, expected_content_type: str = None, response_headers: Dict[str, str] = None):
signing_key = self.get_signing_key()
client = AsyncAppStoreServerAPIClient(signing_key, 'keyId', 'issuerId', 'com.example', Environment.LOCAL_TESTING)
async def fake_execute_and_validate_inputs(method: bytes, url: str, params: Dict[str, Union[str, List[str]]], headers: Dict[str, str], json: Dict[str, Any], data: bytes):
Expand All @@ -754,7 +814,10 @@ async def fake_execute_and_validate_inputs(method: bytes, url: str, params: Dict
self.assertEqual(['User-Agent', 'Authorization', 'Accept'], list(headers.keys()))
self.assertEqual(expected_json, json)

response = Response(status_code, headers={'Content-Type': 'application/json'}, content=body)
response_headers_dict = {'Content-Type': 'application/json'}
if response_headers:
response_headers_dict.update(response_headers)
response = Response(status_code, headers=response_headers_dict, content=body)
return response

client._execute_request = fake_execute_and_validate_inputs
Expand Down