Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
37 changes: 35 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -185,17 +185,50 @@ 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({
'productId': product_id,
}, 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
Expand Down
1 change: 1 addition & 0 deletions example.py
Original file line number Diff line number Diff line change
Expand Up @@ -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}")

Expand Down
26 changes: 17 additions & 9 deletions keymint/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down Expand Up @@ -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.
Expand All @@ -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)
Expand All @@ -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)
Expand Down
13 changes: 12 additions & 1 deletion keymint/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
80 changes: 80 additions & 0 deletions tests/test_contracts.py
Original file line number Diff line number Diff line change
Expand Up @@ -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", "<html>", 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"
Loading