Provides webhook endpoint management (CRUD) and signature verification.
WEBHOOK_EVENT_FAILED module-attribute
WEBHOOK_EVENT_FAILED: str = 'job.failed'Event name sent when a job fails.
WEBHOOK_EVENT_RUNNING module-attribute
WEBHOOK_EVENT_RUNNING: str = 'job.running'Event name sent when a job starts running.
WEBHOOK_EVENT_SUCCEEDED module-attribute
WEBHOOK_EVENT_SUCCEEDED: str = 'job.succeeded'Event name sent when a job succeeds.
WebhooksServiceClient
Bases: ScrubbedSessionState
Client for managing webhook endpoints and verifying signatures.
Manages its own requests.Session and implements the context-manager protocol for clean session teardown.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
api_key | str or _SharedApiKey | API key for authenticating requests. | required |
logger | Logger | Logger instance for debug/info output. | required |
base_url | str | Base URL for the Infrared API (e.g. "https://api.infrared.city/v2"). | required |
close
close() -> NoneClose the underlying requests session.
register
register(url: str, type: str, ip: Optional[str] = None) -> WebhookEndpoint
Register a new webhook endpoint.
POST {base_url}/webhooks with the registration payload.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url | str | The URL to receive webhook events. | required |
type | str | Endpoint type ("production" or "development"). | required |
ip | str or None | Optional IP address filter. | None |
Returns:
| Type | Description |
|---|---|
WebhookEndpoint | The newly created endpoint (without timestamps). |
Raises:
| Type | Description |
|---|---|
WebhookRegistrationError | If the HTTP request fails. |
list
list() -> List[WebhookEndpoint]
List all webhook endpoints.
GET {base_url}/webhooks.
Returns:
| Type | Description |
|---|---|
[list](#infrared_sdk.webhooks.WebhooksServiceClient.list)[[WebhookEndpoint](#infrared_sdk.webhooks.WebhookEndpoint)] | All registered endpoints (with timestamps). |
Raises:
| Type | Description |
|---|---|
WebhookError | If the HTTP request fails. |
get
get(endpoint_id: str) -> WebhookEndpoint
Get a specific webhook endpoint by ID.
GET {base_url}/webhooks/{endpoint_id}.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
endpoint_id | str | The ID of the endpoint to retrieve. | required |
Returns:
| Type | Description |
|---|---|
WebhookEndpoint | The requested endpoint (with timestamps). |
Raises:
| Type | Description |
|---|---|
WebhookNotFoundError | If the endpoint is not found (404). |
WebhookError | If the HTTP request fails for other reasons. |
delete
delete(endpoint_id: str) -> NoneDelete a webhook endpoint by ID.
DELETE {base_url}/webhooks/{endpoint_id}.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
endpoint_id | str | The ID of the endpoint to delete. | required |
Raises:
| Type | Description |
|---|---|
WebhookNotFoundError | If the endpoint is not found (404). |
WebhookError | If the HTTP request fails for other reasons. |
verify_signature staticmethod
verify_signature(
payload_body: bytes, headers: dict, secret: str, tolerance: int = 300
) -> boolVerify a Standard Webhooks HMAC-SHA256 signature.
Validates that the signature in the webhook-signature header matches the expected HMAC for the given payload, and that the timestamp is within the tolerance window (replay protection).
Signing format:
signed_content = "{webhook-id}.{webhook-timestamp}.{body}" signature = "v1," + base64(HMAC-SHA256(decoded_secret, signed_content))
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payload_body | bytes | The raw request body bytes. | required |
headers | dict | Request headers containing webhook-id, webhook-timestamp, and webhook-signature. | required |
secret | str | The webhook secret ("whsec_" prefixed, base64-encoded). | required |
tolerance | int | Maximum age of the timestamp in seconds (default 300). Set to 0 to disable replay protection. | 300 |
Returns:
| Type | Description |
|---|---|
bool | True if the signature is valid and the timestamp is within tolerance; False otherwise. |
WebhookEndpoint dataclass
Immutable snapshot of a webhook endpoint from the API.
The created_at and updated_at fields are optional because the registration response (POST /webhooks) only returns {id, url, type} without timestamps, whereas list/get responses include them.
Attributes:
| Name | Type | Description |
|---|---|---|
id | str | Identifier of the endpoint. |
url | str | The URL that receives webhook events. |
type | str | Endpoint type ("production" or "development"). |
created_at | str or None | Creation time as returned by the API, or None when the response carried no timestamps. |
updated_at | str or None | Last-update time as returned by the API, or None when the response carried no timestamps. |
from_response classmethod
from_response(data: dict) -> WebhookEndpoint
Build a WebhookEndpoint from an API response dict.
Handles both registration responses (no timestamps) and list/get responses (with timestamps via camelCase keys).
WebhookError
Bases: Exception
Base exception for all webhook errors.
HTTP error context lives on attributes (status_code, response_body) instead of the positional message so the raw response body never lands in args[0] / default repr / Sentry's message capture.
Attributes:
| Name | Type | Description |
|---|---|---|
status_code | int or None | HTTP status of the failed response, if there was one. |
response_body | str or None | The raw response text, if any. |
WebhookNotFoundError
Bases: WebhookError
Raised when a webhook endpoint is not found.
WebhookRegistration dataclass
Registration request payload for creating a webhook endpoint.
Attributes:
| Name | Type | Description |
|---|---|---|
url | str | The URL to receive webhook events. |
type | str | Endpoint type ("production" or "development"). |
ip | str or None | Optional IP address filter for the endpoint. |
WebhookRegistrationError
Bases: WebhookError
Raised when webhook registration fails.