Function construct_event

sdks/python/src/e2a/v1/webhook_signature.py:104–137

Verify a delivery and parse it into a :class:`WebhookEvent` in one call (Stripe's ``construct_event`` shape). Raises :class:`~e2a.v1.errors.E2AWebhookSignatureError` on a bad signature, a replay outside tolerance, or an unparseable body. Recommended path — call it from your webhook h

(
    raw_body: Union[str, bytes],
    header: str,
    secret: Secret,
    *,
    tolerance_seconds: int = 300,
    now: Optional[float] = None,
)

def construct_event(
    raw_body: Union[str, bytes],
    header: str,
    secret: Secret,
    *,
    tolerance_seconds: int = 300,
    now: Optional[float] = None,
) -> WebhookEvent:
    """Verify a delivery and parse it into a :class:`WebhookEvent` in one call
    (Stripe's ``construct_event`` shape). Raises
    :class:`~e2a.v1.errors.E2AWebhookSignatureError` on a bad signature, a replay
    outside tolerance, or an unparseable body. Recommended path — call it from
    your webhook handler with the RAW request body.
    """
    if not verify_webhook_signature(
        raw_body, header, secret, tolerance_seconds=tolerance_seconds, now=now
    ):
        raise _sig_error("webhook_signature_invalid", "webhook signature verification failed")

    text = raw_body.decode("utf-8") if isinstance(raw_body, bytes) else raw_body
    try:
        parsed = json.loads(text)
    except (ValueError, TypeError):
        raise _sig_error("webhook_body_invalid", "webhook body is not valid JSON")
    if not isinstance(parsed, dict) or not isinstance(parsed.get("type"), str):
        raise _sig_error("webhook_body_invalid", "webhook event is missing a string `type`")

    return WebhookEvent(
        type=parsed["type"],
        data=parsed.get("data"),
        id=parsed.get("id"),
        created_at=parsed.get("created_at"),
        raw=parsed,
    )


def _sig_error(code: str, message: str) -> E2AWebhookSignatureError: