Events

Events are the immutable, append-only record of every state change in your Plugipay workspace — plugipay.customer.created.v1, plugipay.checkout_session.completed.v1, plugipay.invoice.paid.v1, and so on. Webhooks are how Plugipay pushes events to you; this resource is how you pull them — for replay, audit, debugging missed deliveries, or building a one-time backfill. The Python SDK exposes two methods behind plugipay.events. For HTTP detail and the full event-type catalog, see API → Events.

Namespace

plug.events         # _Events

Methods

list

plug.events.list(
    *,
    limit: int | None = None,
    cursor: str | None = None,
    order: str | None = None,
    type: str | None = None,
    occurred_after: str | None = None,
    occurred_before: str | None = None,
) -> PageResult[EventRecord]

Cursor-paginated list of events.

  • type — filter to a single event type (e.g. "plugipay.checkout_session.completed.v1"). Supports exact match only — no wildcards.
  • order"asc" (oldest first, the default for replay flows) or "desc".
  • occurred_after / occurred_before — ISO 8601 UTC strings, half-open [after, before).
# Replay all checkout completions from May 1 onwards
cursor = None
while True:
    page = plug.events.list(
        limit=100,
        order="asc",
        type="plugipay.checkout_session.completed.v1",
        occurred_after="2026-05-01T00:00:00Z",
        cursor=cursor,
    )
    for event in page.data:
        handle_completed(event)
    if not page.has_more:
        break
    cursor = page.cursor

get

plug.events.get(event_id: str) -> EventRecord

Retrieves one event by id. Useful when a webhook delivery arrived but you've lost the payload, or when a support ticket cites an evt_… id.

event = plug.events.get("evt_01HX...")
print(event["type"], event["occurredAt"])
print(event["data"]["object"])     # the resource snapshot

Pull for replay, push for live. Webhooks deliver every event in near-real-time with at-least-once semantics. The events API is the right tool for one-time backfills, debugging missing webhook deliveries, and building audit views — not the primary integration path. See Webhooks for the push side.

Types

from plugipay import EventRecord, PageResult

EventRecord is a Resource subclass. The shape mirrors the webhook envelope:

  • event["id"]evt_ + ULID.
  • event["type"] — e.g. "plugipay.invoice.paid.v1". The version suffix is part of the type — Plugipay never changes the shape under a published type; it adds .v2.
  • event["accountId"]acc_ + ULID, the owning workspace.
  • event["occurredAt"] — ISO 8601 UTC of when the underlying state change happened.
  • event["data"]["object"] — snapshot of the resource at the moment of the event. A snapshot, not a live pointer — the resource may have moved on since.
  • event.get("livemode")True for live keys, False for test keys.

If you've already parsed it as a WebhookEvent via verify_webhook, the same accessors (event.type, event.id, event.object) work. For events fetched here, you'd use dict access (event["type"], event["data"]["object"]) since EventRecord doesn't include WebhookEvent's convenience properties.

Full reference at API → Events.

Common patterns

Backfill after a webhook outage. Your endpoint was down for two hours; Plugipay's retry queue eventually catches up, but if you want to be sure:

from datetime import datetime, timezone

outage_start = "2026-05-12T14:00:00Z"
outage_end   = "2026-05-12T16:30:00Z"

cursor = None
while True:
    page = plug.events.list(
        limit=100,
        order="asc",
        occurred_after=outage_start,
        occurred_before=outage_end,
        cursor=cursor,
    )
    for event in page.data:
        if not already_processed(event["id"]):
            dispatch(event)
    if not page.has_more:
        break
    cursor = page.cursor

Verify webhook delivery completeness. Cron a sweep that compares delivered ids (your DB) vs Plugipay's record:

from datetime import datetime, timedelta, timezone

since = (datetime.now(timezone.utc) - timedelta(hours=24)).isoformat()
plugipay_ids = set()
cursor = None
while True:
    page = plug.events.list(limit=100, occurred_after=since, cursor=cursor)
    plugipay_ids.update(e["id"] for e in page.data)
    if not page.has_more:
        break
    cursor = page.cursor

missing = plugipay_ids - local_db.delivered_event_ids(since)
for evt_id in missing:
    event = plug.events.get(evt_id)
    dispatch(event)

Idempotent replay. Events are immutable — replaying the same evt_ id is always safe on your side if you guard handlers on event["id"]:

def handle(event):
    if local_db.has_processed(event["id"]):
        return
    with local_db.transaction():
        match event["type"]:
            case "plugipay.invoice.paid.v1":
                mark_paid(event["data"]["object"]["id"])
            case "plugipay.refund.succeeded.v1":
                mark_refunded(event["data"]["object"]["id"])
        local_db.record_processed(event["id"])

Time-bounded audit views. Show "everything that happened to this customer last month":

events_for_customer = []
cursor = None
while True:
    page = plug.events.list(
        limit=100,
        occurred_after="2026-04-01T00:00:00Z",
        occurred_before="2026-05-01T00:00:00Z",
        cursor=cursor,
    )
    for e in page.data:
        obj = e["data"]["object"]
        if obj.get("customerId") == "cus_01HX..." or obj.get("id") == "cus_01HX...":
            events_for_customer.append(e)
    if not page.has_more:
        break
    cursor = page.cursor

(For large workspaces, prefer per-resource list endpoints — plug.invoices.list(customer_id=...) etc. — and stitch from there. Filtering events client-side scales poorly.)

Errors

err.status err.code Cause
400 validation_error Bad type, malformed occurred_after / occurred_before.
404 not_found Event id doesn't exist in this workspace. Events are workspace-scoped.
403 insufficient_scope Key lacks plugipay:event:read.

Read-only — no state-transition errors. See Errors.

Next

Plugipay — Payments that don't tax your success