Developer9 min read

Python SDK

Server-side Python SDK for the Donations Embed API. Async-first, Pydantic v2, FastAPI-friendly. Python 3.10+.

Python SDK

The Python SDK is a typed, async-first client for the Donations Embed Customer SDK API plus a constant-time HMAC-SHA256 webhook verifier. It is built on httpx and pydantic v2, runs on Python 3.10 or newer, and integrates naturally with FastAPI, Starlette, Django (async views), and any other Python web stack.

Install

pip install onefirstflock-donations-embed
# or
uv add onefirstflock-donations-embed

The package name on PyPI is onefirstflock-donations-embed; the import name is donations_embed.

v1.0.0 install bug — install from source until v1.0.1 ships. The published v1.0.0 wheel on PyPI omits donations_embed/models/webhook_test.py and raises ModuleNotFoundError at import time. Until v1.0.1, install the tagged source instead:
pip install "git+https://gitlab.com/1st-flock/donations@v1.0.0#subdirectory=python/donations-embed"
# or
uv add "git+https://gitlab.com/1st-flock/donations@v1.0.0#subdirectory=python/donations-embed"
Tracked in the Changelog → Known issues.

Initialise the client

import os
from donations_embed import Client

client = Client(api_key=os.environ["FLOCK_SECRET_KEY"])

The constructor validates the key prefix; passing a publishable key (pk_…) or a malformed string raises ValueError with a clear message.

Client is intended for use as an async context manager so connection pools are cleaned up on exit:

async with Client(api_key=os.environ["FLOCK_SECRET_KEY"]) as client:
    page = await client.list_donations(limit=25)
    ...

For long-lived applications (FastAPI lifespan, Django async middleware), construct the client once at startup and close it on shutdown:

client = Client(api_key=os.environ["FLOCK_SECRET_KEY"])

@app.on_event("startup")
async def startup() -> None:
    app.state.flock = client

@app.on_event("shutdown")
async def shutdown() -> None:
    await app.state.flock.aclose()

Constructor arguments

Argument	Type	Default	Notes
`api_key`	`str`	(required)	`sk_live_…` or `sk_test_…`. Publishable keys raise `ValueError`.
`base_url`	`str`	`https://api.1stflock.com`	Override for the sandbox cluster (`https://api.sandbox.1stflock.com`) or for self-hosted environments.
`max_retries`	`int`	`3`	Retries on 5xx + transient network errors.
`timeout`	`float`	`30.0`	Per-request timeout in seconds.
`http_client`	`httpx.AsyncClient`	(auto)	Pre-configured client for connection pooling, proxies, or test transports. The SDK does not close a caller-supplied client.

Sandbox

Pass a sandbox key — the SDK routes to the sandbox cluster automatically based on the _test_ prefix.

client = Client(api_key=os.environ["FLOCK_SECRET_KEY"])  # sk_test_…

If you ever need to override the base URL explicitly:

client = Client(
    api_key=os.environ["FLOCK_SECRET_KEY"],
    base_url="https://api.sandbox.1stflock.com",
)

List donations

Cursor-paginated. Page size caps at 200 with a default of 50.

async def stream_all_donations() -> None:
    async with Client(api_key=os.environ["FLOCK_SECRET_KEY"]) as client:
        cursor: str | None = None
        while True:
            page = await client.list_donations(cursor=cursor, limit=200)
            for donation in page.donations:
                print(donation.confirmation_id, donation.gross_amount_cents, donation.status)
            if page.next_cursor is None:
                break
            cursor = page.next_cursor

Filter by fund or donor email:

building_only = await client.list_donations(fund_id="6f1b9b54-0f0b-4d1d-9c8a-9a7e5d4d0001")
sams_history = await client.list_donations(donor_email="sam@example.com")

Get a single donation

donation = await client.get_donation("8a1b9c4d-1234-4321-9abc-def012345678")
print(donation.confirmation_id, donation.donor_email)

Returns 404 (and the SDK raises NotFoundError) if the donation is not visible to the caller, including cross-organisation requests.

Refund a donation

Idempotent on Idempotency-Key. Replays return the original record without re-contacting the gateway.

from donations_embed import Client, idempotency_key

async def refund_full(donation_id: str) -> None:
    async with Client(api_key=os.environ["FLOCK_SECRET_KEY"]) as client:
        result = await client.refund_donation(
            donation_id,
            idempotency_key=idempotency_key("refund"),
        )
        print(result.refund_amount_cents, result.processor_refund_id)

async def refund_partial(donation_id: str) -> None:
    async with Client(api_key=os.environ["FLOCK_SECRET_KEY"]) as client:
        result = await client.refund_donation(
            donation_id,
            amount_cents=1000,                 # $10.00 partial refund
            reason="Donor requested adjustment",
            idempotency_key=idempotency_key("refund"),
        )
        print(result.refund_amount_cents, result.processor_refund_id)

If the network call times out, retry the same call with the same idempotency_key — the server returns the original record without contacting the gateway again.

List recurring schedules

page = await client.list_recurring(status="active", limit=100)
for schedule in page.recurring:
    print(schedule.donor_email, schedule.amount_cents, schedule.interval)

Cancel a recurring schedule

Idempotent. A second call returns already_cancelled=True without re-emitting a webhook.

result = await client.cancel_recurring(
    "5b2c8e10-1ed0-4f0c-bb8a-1234567890ab",
    idempotency_key=idempotency_key("cancel-recurring"),
)
print(result.status, result.already_cancelled)

Fire a synthetic webhook

Useful for end-to-end testing of your webhook handler without making a real donation.

result = await client.test_webhook(event_type="donation.completed")
print("Queued event id:", result.event_id)

Your endpoint receives a normal-looking event whose data.object.synthetic flag is set to True so your handler can branch on it and skip downstream processing during integration testing.

Verify a webhook

The verifier is constant-time and supports both sync and async APIs. It returns a parsed, discriminated event union on success; raises one of WebhookFormatError, WebhookTimestampError, or WebhookSignatureError on failure.

Pass the raw bytes the platform delivered. Re-serialising the JSON before verifying changes the bytes and the signature will not match.

FastAPI

import os
from fastapi import FastAPI, Header, HTTPException, Request
from donations_embed import (
    DonationCompletedEvent,
    DonationRefundedEvent,
    KeyFrozenEvent,
    WebhookFormatError,
    WebhookSignatureError,
    WebhookTimestampError,
    verify_webhook,
)

app = FastAPI()
WEBHOOK_SECRET = os.environ["FLOCK_WEBHOOK_SECRET"]

@app.post("/webhooks/flock")
async def receive_webhook(
    request: Request,
    flock_signature: str = Header(..., alias="Flock-Signature"),
) -> dict[str, str]:
    body = await request.body()
    try:
        event = verify_webhook(body, flock_signature, WEBHOOK_SECRET)
    except (WebhookSignatureError, WebhookTimestampError, WebhookFormatError):
        # Always return 401 — never confirm which check failed.
        raise HTTPException(status_code=401, detail="invalid signature") from None

    match event:
        case DonationCompletedEvent():
            await persist_receipt(event.data.object)
        case DonationRefundedEvent():
            await reverse_receipt(event.data.object)
        case KeyFrozenEvent():
            await alert_on_call(event.data.object)
        case _:
            pass  # Ignore unknown event types.

    # Always 2xx after a successful verify; the platform retries on non-2xx.
    return {"status": "ok"}

Django (async view)

import os
from django.http import HttpRequest, HttpResponse, JsonResponse
from donations_embed import (
    WebhookFormatError,
    WebhookSignatureError,
    WebhookTimestampError,
    verify_webhook_async,
)

WEBHOOK_SECRET = os.environ["FLOCK_WEBHOOK_SECRET"]

async def flock_webhook(request: HttpRequest) -> HttpResponse:
    if request.method != "POST":
        return HttpResponse(status=405)
    signature = request.headers.get("Flock-Signature", "")
    body = request.body  # raw bytes — Django gives them to you as-is

    try:
        event = await verify_webhook_async(body, signature, WEBHOOK_SECRET)
    except (WebhookFormatError, WebhookSignatureError, WebhookTimestampError):
        return HttpResponse("invalid signature", status=401)

    # Dispatch on event.type; persist event.id for at-least-once dedup
    if event.type == "donation.completed":
        await persist_receipt(event.data.object)

    return JsonResponse({"status": "ok"})

Sync vs async verification

from donations_embed import verify_webhook, verify_webhook_async

# Sync — use from sync handlers, worker threads, scripts
event = verify_webhook(body, signature_header, secret)

# Async — use from async middleware
event = await verify_webhook_async(body, signature_header, secret)

Both functions are pure, in-process, and have no network I/O.

Dispatching by event type

Use match (Python 3.10+) or isinstance to dispatch on the discriminated union:

from donations_embed import (
    DonationCreatedEvent,
    DonationCompletedEvent,
    DonationFailedEvent,
    DonationRefundedEvent,
    RecurringActivatedEvent,
    RecurringPaymentSucceededEvent,
    RecurringPaymentFailedEvent,
    RecurringCancelledEvent,
    KeyFrozenEvent,
)

match event:
    case DonationCompletedEvent():
        await persist_receipt(event.data.object)
    case DonationRefundedEvent():
        await reverse_receipt(event.data.object)
    case RecurringActivatedEvent():
        await record_new_subscription(event.data.object)
    case KeyFrozenEvent():
        await alert_on_call(event.data.object)
    case _:
        pass  # ignore unhandled types

Dedup by `Flock-Event-Id`

Webhook delivery is at-least-once. Persist event.id (also exposed as the Flock-Event-Id HTTP header) and short-circuit re-deliveries:

if await already_processed(event.id):
    return {"status": "ok"}
await record_event(event.id)

Error handling

Every API call raises a typed exception on a non-2xx response. Catch the common base (ApiError) for blanket handling, or pivot on specific subclasses for domain logic.

import asyncio
from donations_embed import (
    ApiError,
    AuthError,
    ConflictError,
    ForbiddenError,
    NotFoundError,
    RateLimitError,
    ValidationError,
)

try:
    refund = await client.refund_donation(donation_id, amount_cents=1000)
except RateLimitError as exc:
    await asyncio.sleep(exc.retry_after or 1.0)
    # ... retry
except ConflictError:
    log.warning("Donation %s already fully refunded", donation_id)
except (NotFoundError, AuthError, ForbiddenError):
    raise
except ApiError as exc:
    log.error(
        "Refund failed: status=%s code=%s message=%s request_id=%s",
        exc.status_code, exc.code, exc.message, exc.request_id,
    )
    raise

The full hierarchy:

HTTP	Exception	When
400 / 422	`ValidationError`	Request body or query parameters fail schema validation.
401	`AuthError`	Missing, malformed, or revoked secret key.
403	`ForbiddenError`	Key pair is frozen, or the caller lacks the required RBAC scope.
404	`NotFoundError`	Resource does not exist or is not visible to the caller.
409	`ConflictError`	Conflicting state (e.g. refund of an already-refunded donation).
410	`GoneError`	Session expired or terminally failed.
429	`RateLimitError`	Per-(key, IP) rate limit exceeded. Carries `retry_after`.
502	`BadGatewayError`	Upstream payment gateway error.
503	`ServiceUnavailableError`	Required platform configuration missing.
Other	`ApiError`	Catch-all base class.

Every exception carries:

status_code: int — HTTP status
code: str | None — stable, machine-readable identifier from the response body
message: str — human-readable description
details: dict | None — optional extra data
request_id: str | None — server correlation header (x-request-id)

RateLimitError additionally exposes retry_after: float | None parsed from the Retry-After response header.

Retry behaviour and idempotency

The transport retries 5xx responses (500, 502, 503, 504) and transient network errors (httpx.ConnectError, timeouts) with exponential backoff and jitter. Default schedule: 3 retries with backoff factors of approximately 1 s, 2 s, 4 s plus jitter.

4xx responses are never auto-retried — they are caller errors that retrying will not fix. 429 responses are never auto-retried; the SDK raises RateLimitError with the Retry-After value parsed into retry_after.

Mutating endpoints (refund_donation, cancel_recurring) accept an idempotency_key argument that is forwarded as the Idempotency-Key HTTP header. Replays return the original response without performing the side effect a second time:

from donations_embed import idempotency_key

key = idempotency_key("refund")  # → "refund:1f2dab36-2a55-4f11-bdc8-2a2a4a95b001"
await client.refund_donation(donation_id, amount_cents=1000, idempotency_key=key)

The tag prefix groups related retries under a single observability label so log aggregation tools can collapse them.

Sharing an HTTP client

Pass a pre-configured httpx.AsyncClient to share connection pools, set a proxy, or inject test transports:

import httpx
from donations_embed import Client

shared_http = httpx.AsyncClient(proxies="http://corp-proxy:3128")
client = Client(
    api_key=os.environ["FLOCK_SECRET_KEY"],
    http_client=shared_http,
)

The SDK does not close a caller-supplied client; you own its lifecycle.

Examples

The SDK ships three runnable examples in the examples/ directory of the source repository:

verify_webhook.py — minimal FastAPI endpoint that verifies and dispatches.
refund_donation.py — CLI script that issues a partial refund with an idempotency key.
stream_recurring.py — paginates the entire recurring schedule list and cancels inactive ones.

Versioning

The SDK follows semantic versioning. The 1.x line guarantees backwards compatibility for the public surface — anything imported from the top-level donations_embed package. Internal modules prefixed with an underscore (_http, _idempotency) are not part of the public API and may change between minor versions.