Python SDK — webhooks

Paylera POSTs typed events to your endpoint with a Paylera-Signature header. The signature is HMAC-SHA-256 over <timestamp>.<raw body>, hex-encoded, with multi-v1 rotation support per /webhooks/signatures/.

paylera_fastapi ships three layers — pick the one that fits:

Router factory — paylera_webhooks(...) / paylera_webhook_urls(...) verifies + dispatches in one mount.
Typed event models — PayleraWebhookEvent and its subclasses.
Low-level verifier — verify_signature(...) for everything else.

Router factory (FastAPI)

import os
from paylera_fastapi import paylera_webhooks, InvoicePaidEvent, PayleraWebhookEvent


async def handle_invoice_paid(event: InvoicePaidEvent) -> None:
    # event.data is the typed InvoicePaidEventData
    print(event.data.id, event.data.amount_paid, event.data.currency)


async def handle_fallback(event: PayleraWebhookEvent) -> None:
    # Anything that didn't match a more specific handler.
    print("unhandled", event.type, event.id)


app.include_router(
    paylera_webhooks(
        signing_secret=os.environ["PAYLERA_WEBHOOK_SECRET"],
        accepted_secrets=[os.environ.get("PAYLERA_WEBHOOK_SECRET_PREVIOUS", "")] or None,
        tolerance_seconds=300,
        handlers={
            "invoice.paid":          handle_invoice_paid,
            "subscription.canceled": handle_subscription_canceled,
            "invoice.*":             handle_invoice_paid,    # prefix wildcard
            "*":                     handle_fallback,        # global fallback
        },
    ),
    prefix="/webhooks/paylera",
)

What the router does on every POST:

Reads the raw body — never request.json(), which would re-serialise and break the HMAC.
Verifies the Paylera-Signature header against the configured secret(s) within the 5-minute replay window.
Parses the envelope into the most specific PayleraWebhookEvent subclass for the type.
Dispatches to the handler registered for that type (or prefix.*, or * as the global fallback).

Response codes:

Status	When
`200`	Handler ran, or no handler matched. (4xx means “don’t retry”.)
`401`	Missing / malformed / mismatched signature; replay-window violation.
`500`	Registered handler raised. Paylera retries with backoff.

Router factory (Django)

from paylera_django import paylera_webhook_urls
from paylera_fastapi import InvoicePaidEvent


async def handle_invoice_paid(event: InvoicePaidEvent) -> None:
    ...


urlpatterns = [
    path("webhooks/paylera/", include(paylera_webhook_urls(
        signing_secret=settings.PAYLERA_WEBHOOK_SECRET,
        handlers={"invoice.paid": handle_invoice_paid},
    ))),
]

Same verification + dispatch contract as the FastAPI version — the two adapters share the underlying WebhookCore.

Typed event models

Typed Pydantic models ship for the high-value families:

Type	Model
`invoice.created`	`InvoiceCreatedEvent`
`invoice.paid`	`InvoicePaidEvent`
`invoice.payment_failed`	`InvoicePaymentFailedEvent`
`subscription.created`	`SubscriptionCreatedEvent`
`subscription.updated`	`SubscriptionUpdatedEvent`
`subscription.canceled`	`SubscriptionCanceledEvent`
`payment.succeeded`	`PaymentSucceededEvent`
`payment.failed`	`PaymentFailedEvent`
`customer.external_ref_discovered`	`CustomerExternalRefDiscoveredEvent`
`revenue.captured`	`RevenueCapturedEvent`
`revenue.refunded`	`RevenueRefundedEvent`
`revenue.subscription_started`	`RevenueSubscriptionStartedEvent`
`revenue.subscription_renewed`	`RevenueSubscriptionRenewedEvent`
`revenue.subscription_canceled`	`RevenueSubscriptionCanceledEvent`
`revenue.subscription_grace_period_entered`	`RevenueSubscriptionGracePeriodEnteredEvent`
`revenue.subscription_recovered`	`RevenueSubscriptionRecoveredEvent`
`revenue.subscription_expired`	`RevenueSubscriptionExpiredEvent`
`revenue.one_time_purchase`	`RevenueOneTimePurchaseEvent`
`revenue.unknown_currency_seen`	`RevenueUnknownCurrencySeenEvent`
(everything else)	`PayleraWebhookEvent` (generic, `data` is `dict`)

The 10 revenue.* models share a RevenueData family in paylera_fastapi.events. They are observability projections from Stripe / Toss / Apple App Store Server Notifications V2 / Google Play RTDN — see Multi-source revenue capture for the family overview and the money-bearing vs. lifecycle split. CustomerExternalRefDiscoveredEvent is emitted by the RevenueAttribution worker when a previously-orphaned revenue.* event is back-linked to an existing customer via email-fingerprint match.

Each event carries the standard envelope fields — id, type, created_at, tenant_id, api_version, data, idempotency_key, request_id. Handler signatures can use either the specific or the generic type:

async def handle_payment(event: PaymentSucceededEvent) -> None: ...
async def handle_anything(event: PayleraWebhookEvent) -> None: ...

Secret rotation

Pass the in-flight secret as signing_secret= and the about-to-be-retired or newly-deployed secret as accepted_secrets=[...]. The verifier accepts any match across the set, so a single rotation window can deploy in either order:

paylera_webhooks(
    signing_secret=os.environ["PAYLERA_WEBHOOK_SECRET"],
    accepted_secrets=[
        os.environ.get("PAYLERA_WEBHOOK_SECRET_PREVIOUS", ""),
        os.environ.get("PAYLERA_WEBHOOK_SECRET_NEXT", ""),
    ],
)

Low-level verifier

For custom frameworks or one-off scripts:

from paylera_fastapi import verify_signature, SignatureVerificationError

try:
    verify_signature(
        raw_body_bytes,                           # bytes, untouched
        request.headers["Paylera-Signature"],
        signing_secret=os.environ["PAYLERA_WEBHOOK_SECRET"],
        tolerance_seconds=300,
        strict=True,                              # raise instead of returning False
    )
except SignatureVerificationError as err:
    log.warning("paylera signature rejected: %s", err.reason)
    return Response(status_code=401)

verify_signature(strict=False) (the default) returns True / False instead. The exception’s reason attribute is one of: missing_header, malformed_header, clock_skew, signature_mismatch, replay_too_old.

Error-handler hook

Both factories take an optional on_error= callback fired when a handler raises. Useful for shipping the exception to Sentry without interfering with FastAPI/Django’s own logging:

def report(exc: BaseException, event: PayleraWebhookEvent | None) -> None:
    sentry_sdk.capture_exception(exc, extra={"event_id": event.id if event else None})


paylera_webhooks(..., on_error=report)

Local testing

Use ngrok or any inbound tunnel to forward Paylera’s deliveries to your laptop. The webhooks/local-testing guide covers forwarding, replaying, and inspecting deliveries.