pulsepigeon

Official PulsePigeon API client for Python.

Install

pip install pulsepigeon

Quickstart

import os
from pulsepigeon_client import PulsePigeonClient

client = PulsePigeonClient(api_key=os.environ["PULSEPIGEON_API_KEY"])

client.send_message(
    {
        "project_id": "proj_123",
        "from_email": "support@example.com",
        "to": [{"email": "user@example.com"}],
        "subject": "Reset your password",
        "text": "Use the secure reset link in this email.",
    },
    idempotency_key="password-reset-user-123-event-456",
)

Every send should carry an idempotency_key — retrying the same key with the same body returns the original result instead of sending a duplicate.

Personalized batches use per-item idempotency and partial-success results:

result = client.send_batch(
    [
        {
            "idempotency_key": "invoice-123-user-456",
            "message": {
                "project_id": "proj_123",
                "from_email": "billing@example.com",
                "to": [{"email": "user@example.com"}],
                "subject": "Your invoice",
                "text": "Invoice 123 is ready.",
            },
        }
    ]
)

The API accepts at most 100 items and 10 MB of decoded JSON per request. Accepted items replay their original message IDs. Failed items return an HTTP-style status and error code in their own result.

An AsyncPulsePigeonClient with the same methods is available for asyncio-based applications (each call runs the sync request on a worker thread — see the module docstring for why that's the right trade-off for a stdlib-only client):

from pulsepigeon_client import AsyncPulsePigeonClient

client = AsyncPulsePigeonClient(api_key=os.environ["PULSEPIGEON_API_KEY"])
await client.send_message({...}, idempotency_key="...")

API

Every method requires an API key carrying the scope listed below (request scopes when creating the key — POST /v1/api-keys, scopes field — a closed vocabulary: messages:send, reports:read, domains:*, webhooks:*, suppressions:*, contacts:*). Your own session's role bounds which scopes you can request: minting a key with contacts:*, suppressions:*, or webhooks:* requires an owner, admin, or operator session (those scopes reach owner-only routes like GDPR erasure and suppression-list export/import); messages:send, reports:read, and domains:* require only the ordinary role permission their routes already need of a session caller.

Project and workspace scoping. A project key is issued against exactly one project. list_messages, search_message_logs, list_events, list_subscriber_lists, and create_subscriber_list are filtered to that project only — a key never sees or creates data in another project. request_subscriber_opt_in additionally checks the target list's own project before dispatching mail, and requires messages:send on top of contacts:* (it sends a real confirmation email — "manage contacts" must not imply "send mail").

Tenant-wide methods require an owner-created workspace key. Project keys continue to reject those calls with 403 api_key_route_not_project_scoped. Workspace keys require an explicit live or test environment and can restrict source IP networks, sender domains, and expiration. The server enforces each restriction on every request.

Errors raise one of PulsePigeonAuthenticationError (401), PulsePigeonPermissionError (403), PulsePigeonNotFoundError (404), PulsePigeonConflictError (409), PulsePigeonValidationError (400/422), PulsePigeonRateLimitError (429, has retry_after_seconds), or PulsePigeonServerError (5xx) — all subclasses of PulsePigeonError (status, body, code). Catch the base class as a fallback.

Verifying webhook deliveries

PulsePigeon-Signature: t=<unix>,v1=<hmac-sha256 hex> is sent with every status-webhook delivery. Verify it before trusting the payload — this is the single most common thing to get wrong by hand: the signature covers a canonical re-serialization of the JSON body (sorted keys, no whitespace), not the raw bytes received over the wire. The signed timestamp is when the event was enqueued for delivery, not when the HTTP request is actually sent — tolerance_seconds (default 3600) has to budget for realistic delivery-queue delay, not just clock skew, or a delayed-but-authentic delivery gets spuriously rejected.

from pulsepigeon_client import verify_webhook_signature

payload = json.loads(raw_body)
if not verify_webhook_signature(payload, signature_header, secret):
    raise ValueError("invalid PulsePigeon webhook signature")

Auth

Create an API key from the PulsePigeon console under API keys (or POST /v1/api-keys with a scopes list) and pass it as api_key. Keys are sent as Authorization: Bearer <key> — never log or commit a real key.

Integration test

tests/test_bearer_auth_integration.py proves the packaged client round-trips against a real, running instance of the API (not mocked) — see that file and its TypeScript counterpart, sdk/typescript/test/integration.test.ts, for what it verifies and its prerequisites (a reachable Postgres). tests/test_client_unit.py covers request-building and error-mapping without a server. Run with pytest -q.