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
eventsAPI 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")—Truefor live keys,Falsefor 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
- API → Events — full event-type catalog and retention policy.
- Webhooks — the push-based delivery channel.
- Webhook endpoints — register where Plugipay pushes events.