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
1 change: 1 addition & 0 deletions seam/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,3 +14,4 @@
SeamActionAttemptTimeoutError,
)
from .seam_webhook import SeamWebhook
from svix.webhooks import WebhookVerificationError as SeamWebhookVerificationError
24 changes: 22 additions & 2 deletions seam/seam_webhook.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,33 @@
from typing import Dict
from svix import Webhook
from svix.webhooks import Webhook
from .routes.models import SeamEvent


class SeamWebhook:
"""Verifies and parses incoming Seam webhook events using the Svix library."""

def __init__(self, secret: str):
"""
:param secret: The secret key used for webhook verification.
:type secret: str
"""
self._webhook = Webhook(secret)

def verify(self, payload: str, headers: Dict[str, str]) -> SeamEvent:
res = self._webhook.verify(payload, headers)
"""Verify and parse an incoming HTTP webhook request.

Normalizes the headers, verifies the payload using the Svix
Webhook instance, and returns a SeamEvent object.

:param payload: The raw HTTP request body.
:type payload: str
:param headers: The HTTP request headers.
:type headers: Dict[str, str]
:return: The SeamEvent object created from the verified payload.
:rtype: SeamEvent
:raises WebhookVerificationError: If the webhook signature verification fails.
"""
normalized_headers = {k.lower(): v for k, v in headers.items()}
res = self._webhook.verify(payload, normalized_headers)

return SeamEvent.from_dict(res)