diff --git a/README.md b/README.md index 0b52681..2994d8a 100644 --- a/README.md +++ b/README.md @@ -185,9 +185,40 @@ Keymint provides utilities to uniquely identify machines for node-locking: |-------------------------|--------------------------------------------------| | `verify_webhook_signature`| Verifies the signature of a webhook request payload. | +## Error Handling + +Every failed call raises `KeyMintApiError` with: + +| Attribute | Meaning | +|---|---| +| `error_code` | Stable catalogue string from the response's `errorCode`, e.g. `api_key_invalid`, `license_not_found`, `activation_limit_reached`, `product_not_found`, `invalid_request`, `payload_too_large`, `rate_limited`, `session_invalid`, `session_expired`. `None` when the response has none (for example a non-JSON body from a proxy, or a network error). | +| `status` | HTTP status, or `None` when no response was received. | +| `code` | Numeric `code` from the response body, or `-1`. | +| `message` | Human-readable message. It can change, so don't branch on it. | + +Branch on `error_code`, falling back to `status`: + +```python +from keymint import KeyMintApiError + +try: + sdk.activate_key({'productId': product_id, 'licenseKey': license_key, 'hostId': host_id}) +except KeyMintApiError as err: + if err.error_code == 'activation_limit_reached': + ... # ask the customer to deactivate another device + elif err.error_code == 'license_not_found': + ... # wrong or expired key (see License Expiry) + elif err.error_code == 'rate_limited' or err.status == 429: + ... # back off and retry later + elif err.status is None or err.status >= 500: + ... # network or server problem: retry, or fall back to a cached result + else: + raise +``` + ## Idempotency -All mutating SDK methods support idempotency keys to safely retry requests in case of network drops. Pass a unique string (recommended: UUID v4) as the optional `idempotency_key` keyword argument: +Mutating SDK methods support idempotency keys to safely retry requests in case of network drops. Pass a unique string (recommended: UUID v4) as the optional `idempotency_key` keyword argument: ```python result = sdk.create_key({ @@ -195,7 +226,9 @@ result = sdk.create_key({ }, idempotency_key='9b1deb4d-3b7d-4bad-9bdd-2b0d7b3dcb6d') ``` -Use a fresh random UUID for each operation and reuse it only to retry that same operation. Never derive it from the license key or host ID: a replay returns the original response, including floating-session secrets. +Use a fresh random UUID for each operation and reuse it only to retry that same operation. Never derive it from the license key or host ID: a replay returns the original response. + +The floating-license methods (`floating_checkout`, `floating_heartbeat`, `floating_checkin`) accept `idempotency_key` for compatibility, but the server ignores `Idempotency-Key` on those endpoints: every call is processed and none is replayed. ## License MIT diff --git a/example.py b/example.py index bee5fcb..4a9f425 100644 --- a/example.py +++ b/example.py @@ -102,6 +102,7 @@ def main(): print(f" Message: {e.message}") print(f" Code: {e.code}") print(f" Status: {e.status}") + print(f" Error code: {e.error_code}") except Exception as e: print(f"\n❌ Unexpected error: {e}") diff --git a/keymint/__init__.py b/keymint/__init__.py index 386fce4..d9d220d 100644 --- a/keymint/__init__.py +++ b/keymint/__init__.py @@ -62,18 +62,23 @@ def _handle_request(self, method: str, endpoint: str, params: dict | None = None except requests.exceptions.HTTPError as http_err: try: error_data = http_err.response.json() - nested_error = error_data.get('error') if isinstance(error_data.get('error'), dict) else {} - raise KeyMintApiError( - message=error_data.get('message') or nested_error.get('message') or 'An API error occurred', - code=error_data.get('code', -1), - status=http_err.response.status_code - ) except ValueError: + # Not JSON (e.g. an HTML page from a proxy). raise KeyMintApiError( message=str(http_err), code=-1, status=http_err.response.status_code ) + if not isinstance(error_data, dict): + error_data = {} + nested_error = error_data.get('error') if isinstance(error_data.get('error'), dict) else {} + error_code = error_data.get('errorCode') + raise KeyMintApiError( + message=error_data.get('message') or nested_error.get('message') or 'An API error occurred', + code=error_data.get('code', -1), + status=http_err.response.status_code, + error_code=error_code if isinstance(error_code, str) else None, + ) except Exception as err: raise KeyMintApiError(message=str(err), code=-1) @@ -117,7 +122,8 @@ def floating_checkout(self, params: FloatingCheckoutParams, idempotency_key: str """ Checks out a floating license seat. :param params: Parameters for checking out the license. - :param idempotency_key: Optional unique identifier to ensure request idempotency. + :param idempotency_key: Accepted for compatibility; the server ignores + ``Idempotency-Key`` on floating endpoints and never replays them. :returns: The checkout response containing sessionId and sessionSecret. ``expiresAt`` is the seat lease; ``expirationDate`` is the license expiry as an ISO 8601 string, or ``None`` when the license never expires. @@ -128,7 +134,8 @@ def floating_heartbeat(self, params: FloatingHeartbeatParams, idempotency_key: s """ Sends a heartbeat to keep a floating license session alive. :param params: Parameters for the heartbeat (includes rotating signature). - :param idempotency_key: Optional unique identifier to ensure request idempotency. + :param idempotency_key: Accepted for compatibility; the server ignores + ``Idempotency-Key`` on floating endpoints and never replays them. :returns: The heartbeat response with extended expiry and new nonce. """ return self._handle_request('POST', '/key/heartbeat', params, idempotency_key=idempotency_key) @@ -137,7 +144,8 @@ def floating_checkin(self, params: FloatingCheckinParams, idempotency_key: str | """ Checks in a floating license session, releasing the seat. :param params: Parameters for checking in the license (includes rotating signature). - :param idempotency_key: Optional unique identifier to ensure request idempotency. + :param idempotency_key: Accepted for compatibility; the server ignores + ``Idempotency-Key`` on floating endpoints and never replays them. :returns: The checkin confirmation. """ return self._handle_request('POST', '/key/checkin', params, idempotency_key=idempotency_key) diff --git a/keymint/types.py b/keymint/types.py index b5d0335..5b654c0 100644 --- a/keymint/types.py +++ b/keymint/types.py @@ -36,11 +36,22 @@ class CreateKeyResponse(TypedDict, total=False): keys: list[str] class KeyMintApiError(Exception): - def __init__(self, message: str, code: int, status: int | None = None): + """An error returned by the Keymint API, or a failure reaching it. + + :ivar message: Human-readable message. Don't branch on it; it can change. + :ivar code: Numeric ``code`` from the response body, or -1. + :ivar status: HTTP status, or None when no response was received. + :ivar error_code: Stable catalogue string from the body's ``errorCode`` + (e.g. ``'license_not_found'``, ``'activation_limit_reached'``, + ``'rate_limited'``), or None when the body has none. + """ + + def __init__(self, message: str, code: int, status: int | None = None, error_code: str | None = None): super().__init__(message) self.message = message self.code = code self.status = status + self.error_code = error_code class ActivateKeyParams(TypedDict): productId: str diff --git a/tests/test_contracts.py b/tests/test_contracts.py index e1c5330..1c3b531 100644 --- a/tests/test_contracts.py +++ b/tests/test_contracts.py @@ -124,3 +124,83 @@ def test_delete_customer_still_uses_query_params_and_no_body(mock_delete): assert args[0] == "https://api.keymint.dev/customer/by-id" assert kwargs["params"] == {"customerId": "cust_123"} assert set(kwargs) == {"params", "headers"} + + +@patch("keymint.requests.post") +def test_error_exposes_error_code(mock_post): + mock_post.return_value = response({ + "message": "Activation limit reached", "code": 2, "errorCode": "activation_limit_reached", + }, 403) + + with pytest.raises(KeyMintApiError, match="Activation limit reached") as excinfo: + KeyMint("client_test").activate_key({"productId": "product_123", "licenseKey": "license_123"}) + + assert excinfo.value.error_code == "activation_limit_reached" + assert excinfo.value.code == 2 + assert excinfo.value.status == 403 + assert str(excinfo.value) == "Activation limit reached" + + +@patch("keymint.requests.post") +def test_error_with_legacy_envelope_and_error_code(mock_post): + mock_post.return_value = response({ + "success": False, + "error": {"code": "CUSTOMER_EMAIL_EXISTS", "message": "Customer email already exists in team"}, + "code": 1, + "errorCode": "invalid_request", + }, 409) + + with pytest.raises(KeyMintApiError) as excinfo: + KeyMint("admin_test").create_key({"productId": "product_123"}) + + assert excinfo.value.error_code == "invalid_request" + assert excinfo.value.message == "Customer email already exists in team" + + +@patch("keymint.requests.post") +def test_error_without_error_code_is_none(mock_post): + mock_post.return_value = response({"message": "Invalid license key", "code": 1}, 404) + + with pytest.raises(KeyMintApiError) as excinfo: + KeyMint("client_test").activate_key({"productId": "product_123", "licenseKey": "license_123"}) + + assert excinfo.value.error_code is None + assert excinfo.value.code == 1 + assert excinfo.value.status == 404 + + +@patch("keymint.requests.post") +def test_error_with_non_json_body(mock_post): + result = response(None, 502) + result.json.side_effect = requests.exceptions.JSONDecodeError("Expecting value", "", 0) + mock_post.return_value = result + + with pytest.raises(KeyMintApiError) as excinfo: + KeyMint("client_test").activate_key({"productId": "product_123", "licenseKey": "license_123"}) + + assert excinfo.value.error_code is None + assert excinfo.value.code == -1 + assert excinfo.value.status == 502 + + +@patch("keymint.requests.post") +def test_error_with_non_object_json_body(mock_post): + mock_post.return_value = response(["unexpected"], 500) + + with pytest.raises(KeyMintApiError) as excinfo: + KeyMint("client_test").activate_key({"productId": "product_123", "licenseKey": "license_123"}) + + assert excinfo.value.error_code is None + assert excinfo.value.status == 500 + + +def test_error_constructor_is_backwards_compatible(): + positional = KeyMintApiError("boom", 3, 429) + assert (positional.message, positional.code, positional.status, positional.error_code) == ("boom", 3, 429, None) + + keyword = KeyMintApiError(message="boom", code=-1) + assert keyword.status is None and keyword.error_code is None + + full = KeyMintApiError("Too many requests", 3, 429, "rate_limited") + assert full.error_code == "rate_limited" + assert str(full) == "Too many requests"