Dramatiq Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance

Dramatiq is a fast, reliable Python background task library built on Redis or RabbitMQ. Its actor decorator, composable middleware stack, and first-class group() and pipeline() primitives make it a popular choice for batch billing workloads. Three framework-specific failure modes emerge when Stripe calls run inside Dramatiq actors — and none of them show up as an error in your worker logs.

This post covers three Dramatiq-specific failure modes — actor retry re-execution, group() fan-out key sharing, and APScheduler duplicate dispatch — each of which produces a second stripe.charges.create() call on an already-billed customer, and the two-layer governance pattern that closes all three: content-hash idempotency keys plus per-batch vault keys via a spend-cap proxy, without changing your actor topology.

Failure mode 1: Retries middleware re-invokes the actor callable from line 1 after stripe.charges.create() has already succeeded

Dramatiq's Retries middleware is included in the default middleware stack and is enabled by default with max_retries=20. When an unhandled exception propagates out of an actor function, the middleware intercepts it, increments a retry counter header on the original message, and re-sends the message to the queue with an exponential backoff delay. The actor function is then called from the beginning with the exact same arguments the message was originally sent with. There is no mid-function resume capability — each retry attempt is a full re-invocation of the actor callable.

The failure window is between stripe.charges.create() returning a successful response and the actor function exiting without error. Consider a billing actor that charges a customer and then writes the charge ID to a PostgreSQL table. If the database connection times out after the Stripe call succeeds — or a network error, a lock contention exception, or any other downstream operation raises — the exception propagates out of the actor. Dramatiq catches it via the Retries middleware, schedules a retry, and re-enqueues the message. On the next attempt, the actor calls stripe.charges.create() again with the same customer_id, amount_cents, and billing_period. Without an idempotency key, Stripe creates ch_B. Both charge objects are valid; the customer has been charged twice; and the Dramatiq worker log shows the actor succeeded on attempt 2 with no indication of the duplicate.

# actors/billing.py — UNSAFE: no idempotency key
import os
import stripe
import dramatiq

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]  # shared, unrestricted

@dramatiq.actor(queue_name="billing", max_retries=3)
def charge_customer(customer_id: str, amount_cents: int, billing_period: str) -> None:
    # If this succeeds but the DB write below raises → actor retried → ch_B created
    charge = stripe.charges.create(
        amount=amount_cents,
        currency="usd",
        customer=customer_id,
        description=f"Subscription {billing_period}",
    )

    # psycopg2.OperationalError here → Retries middleware re-sends message
    with db.connect() as conn:
        conn.execute(
            "INSERT INTO charges (charge_id, customer_id, billing_period)"
            " VALUES (%s, %s, %s)",
            (charge.id, customer_id, billing_period),
        )

The fix is a content-hash idempotency key derived from the actor arguments that Dramatiq holds constant across all retry attempts. Because customer_id, amount_cents, and billing_period are encoded in the message at send time and never modified by the middleware on retry, a SHA-256 hash of these three values is identical on every attempt. Stripe returns the original ch_A on every retry rather than creating ch_B. Permanent Stripe errors — card declined, invalid customer — should be caught and returned silently so the Retries middleware does not continue scheduling retry attempts for a charge that will always fail.

# actors/billing.py — SAFE: stable idempotency key + permanent errors suppressed
import hashlib
import os
import stripe
import dramatiq

@dramatiq.actor(queue_name="billing", max_retries=3)
def charge_customer(customer_id: str, amount_cents: int, billing_period: str,
                    vault_key: str) -> None:
    idempotency_key = hashlib.sha256(
        f"{customer_id}:{amount_cents}:{billing_period}:dramatiq-billing".encode()
    ).hexdigest()[:32]

    # Per-batch vault key from actor args — not os.environ["STRIPE_SECRET_KEY"]
    stripe_client = stripe.Stripe(
        vault_key,
        base_url="https://proxy.keybrake.com/stripe/v1",
    )

    try:
        charge = stripe_client.charges.create(
            amount=amount_cents,
            currency="usd",
            customer=customer_id,
            description=f"Subscription {billing_period}",
            idempotency_key=idempotency_key,
        )
    except stripe.error.CardError as e:
        # Permanent — write declined record and return; Retries will not re-enqueue
        with db.connect() as conn:
            conn.execute(
                "INSERT INTO charges (customer_id, billing_period, status)"
                " VALUES (%s, %s, 'card_declined') ON CONFLICT DO NOTHING",
                (customer_id, billing_period),
            )
        return  # not re-raised → actor succeeds → no further retry
    except stripe.error.InvalidRequestError:
        return  # invalid customer / bad params → permanent, return silently
    # stripe.error.APIConnectionError → re-raise → Retries re-enqueues with backoff
    # idempotency key is stable; Stripe returns ch_A, not ch_B

    with db.connect() as conn:
        conn.execute(
            "INSERT INTO charges (charge_id, customer_id, billing_period, status)"
            " VALUES (%s, %s, %s, 'charged') ON CONFLICT DO NOTHING",
            (charge.id, customer_id, billing_period),
        )

Failure mode 2: group().run() fan-out shares one unrestricted STRIPE_SECRET_KEY across all concurrent actors

Dramatiq's group() primitive dispatches a collection of messages that execute independently and in parallel across the worker pool. When you call dramatiq.group([charge_customer.message(cid, amt, period) for cid, amt in batch]).run(), all messages enter the queue simultaneously. Multiple worker processes — or threads in a multi-threaded worker — dequeue and execute them concurrently. Each worker process reads stripe.api_key from its own copy of the module's global state, which was set from os.environ["STRIPE_SECRET_KEY"] at import time. All concurrent executions therefore share one unrestricted Stripe key with no per-actor spend boundary.

The blast radius of a data error scales directly with your worker concurrency. If amount_cents contains a unit bug — 299900 instead of 2999 — every actor in the group charges its customer the wrong amount before any single actor returns a result. With 500 customers dispatched as a group() and 8 worker processes, 8 incorrect charges fire simultaneously in the first wave. The remaining 492 jobs are already in the queue. There is no circuit breaker at the key layer to halt them once the first errors are visible in monitoring — the unrestricted key authorizes any charge, at any amount, for any customer, without a ceiling.

# dispatch.py — UNSAFE: group() fan-out with one unrestricted key
import dramatiq
from actors.billing import charge_customer

def run_monthly_billing(billing_period: str, customers: list[dict]) -> None:
    # All N actors share one stripe.api_key from os.environ — no per-actor cap
    dramatiq.group([
        charge_customer.message(
            c["id"],
            c["amount_cents"],  # bug here hits every customer simultaneously
            billing_period,
        )
        for c in customers
    ]).run()

The fix is to issue a per-batch vault key before calling group().run(), cap it at the sum of all expected charges multiplied by a 1.10 buffer, and pass it as an argument in each message. The vault key is scoped to POST /v1/charges only and carries a daily spend cap enforced by the proxy across all concurrent actor calls sharing that key. Even if all 8 worker processes dequeue group messages simultaneously, the proxy's total spend cap for the key halts further charges once the cap is reached — the circuit breaker is at the proxy layer rather than requiring application-level coordination across independent worker processes.

# dispatch.py — SAFE: vault key issued once before group dispatch
import hashlib
import os
import requests
import dramatiq
from actors.billing import charge_customer

def run_monthly_billing(billing_period: str, customers: list[dict]) -> None:
    total_cents = sum(c["amount_cents"] for c in customers)
    max_spend_usd = round((total_cents / 100) * 1.10, 2)

    # Issue one vault key covering the entire group
    resp = requests.post(
        "https://proxy.keybrake.com/keys",
        headers={"Authorization": f"Bearer {os.environ['KEYBRAKE_API_KEY']}"},
        json={
            "vendor": "stripe",
            "allowed_endpoints": ["POST /v1/charges"],
            "daily_usd_cap": max_spend_usd,
            "expires_in_seconds": 7200,
            "label": f"dramatiq-billing-{billing_period}",
        },
        timeout=10,
    )
    resp.raise_for_status()
    vault_key = resp.json()["vault_key"]

    dramatiq.group([
        charge_customer.message(
            c["id"],
            c["amount_cents"],
            billing_period,
            vault_key,  # per-batch vault key passed as actor arg
        )
        for c in customers
    ]).run()

Failure mode 3: APScheduler actor.send() and a manual actor.send() create two independent billing executions for the same period

Dramatiq does not include a built-in scheduler. The most common integration pattern is APScheduler firing actor.send(billing_period) on a cron schedule. When actor.send() is called — whether by APScheduler or manually — Dramatiq generates a new UUID for the message's message_id field and enqueues it. If APScheduler fires charge_all_customers.send("2026-07") at 00:00 UTC and an operator, monitoring script, or retry webhook also calls charge_all_customers.send("2026-07") at 00:02 UTC — believing the scheduled run has stalled — two separate messages with different UUIDs exist in the queue. Dramatiq has no concept of deduplication by message arguments; it routes by message_id. Both messages are dequeued independently, both call charge_all_customers("2026-07"), and both dispatch sub-actors to charge every customer in the billing period cohort. Every customer is charged twice.

Unlike some task queues that surface this as an obvious duplicate job record, Dramatiq workers simply execute both messages and report both as succeeded. The Dramatiq admin dashboard, if used, shows two completed messages with different IDs and identical arguments. There is no built-in flag for "duplicate dispatch." The Stripe dashboard will have two charge objects per customer created seconds or minutes apart, both with status: succeeded.

# UNSAFE: APScheduler send + manual send create two independent billing executions
from apscheduler.schedulers.background import BackgroundScheduler
import dramatiq

@dramatiq.actor(queue_name="billing", max_retries=3)
def charge_all_customers(billing_period: str, vault_key: str) -> None:
    customers = db.fetch_unpaid_customers(billing_period)
    dramatiq.group([
        charge_customer.message(c["id"], c["amount_cents"], billing_period, vault_key)
        for c in customers
    ]).run()

scheduler = BackgroundScheduler()
scheduler.add_job(
    charge_all_customers.send,
    "cron",
    hour=0,
    minute=0,
    args=["2026-07", os.environ["STRIPE_SECRET_KEY"]],  # also unsafe: raw key
    # No dedup: APScheduler generates a new message UUID on every fire
)
scheduler.start()

# Operator at 00:02 UTC (assuming stall):
charge_all_customers.send("2026-07", os.environ["STRIPE_SECRET_KEY"])
# → two messages in queue → two workers → N customers × 2 charges each

Two layers close this failure mode. The first is a Redis-based distributed lock using SET NX EX inside the actor, keyed to the billing period. The actor acquires the lock on entry; if the lock already exists (another execution is running or recently completed), the actor logs and returns immediately rather than dispatching the group. The lock TTL is set to the maximum expected batch duration plus a buffer. The second layer is the content-hash idempotency key inside each charge_customer actor, which protects against the race where both billing actors pass the lock check in the same millisecond window before either has written the lock. Together they prevent both the common case (sequential duplicate dispatch) and the uncommon case (concurrent duplicate dispatch).

# SAFE: Redis NX lock guards single execution per billing period
import hashlib
import os
import redis
import requests
import dramatiq

_redis = redis.Redis.from_url(os.environ["REDIS_URL"])

@dramatiq.actor(queue_name="billing", max_retries=1)
def charge_all_customers(billing_period: str) -> None:
    lock_key = f"billing:lock:{billing_period}"
    lock_ttl = 7200  # seconds — covers largest expected batch duration

    # Acquire distributed lock; NX = only set if not exists
    acquired = _redis.set(lock_key, "1", nx=True, ex=lock_ttl)
    if not acquired:
        # Another execution is running or completed for this period — skip silently
        return

    try:
        customers = db.fetch_unpaid_customers(billing_period)
        if not customers:
            return

        total_cents = sum(c["amount_cents"] for c in customers)
        resp = requests.post(
            "https://proxy.keybrake.com/keys",
            headers={"Authorization": f"Bearer {os.environ['KEYBRAKE_API_KEY']}"},
            json={
                "vendor": "stripe",
                "allowed_endpoints": ["POST /v1/charges"],
                "daily_usd_cap": round((total_cents / 100) * 1.10, 2),
                "expires_in_seconds": lock_ttl,
                "label": f"dramatiq-billing-{billing_period}",
            },
            timeout=10,
        )
        resp.raise_for_status()
        vault_key = resp.json()["vault_key"]

        dramatiq.group([
            charge_customer.message(c["id"], c["amount_cents"], billing_period, vault_key)
            for c in customers
        ]).run()
    except Exception:
        # Release lock on unexpected failure so the next scheduled run can proceed
        _redis.delete(lock_key)
        raise  # re-raise → Retries middleware re-enqueues with backoff

# APScheduler fires this; manual send() from operator uses the same actor
# → both hit the Redis NX lock → only one execution proceeds per period
scheduler.add_job(charge_all_customers.send, "cron", hour=0, minute=0, args=["2026-07"])

Approach comparison

Gap analysis

1. @dramatiq.actor(time_limit=N) can kill the actor thread mid-charge, creating an ambiguous charge state

Dramatiq's TimeLimit middleware enforces a wall-clock deadline on actor execution. When the limit is reached it raises dramatiq.middleware.time_limit.TimeLimitExceeded inside the actor thread. If the exception fires after stripe.charges.create() has transmitted the HTTP request to Stripe — but before the Python response object is assigned to a local variable — the charge exists on Stripe's side with no record locally, and the actor dies with TimeLimitExceeded. Dramatiq's Retries middleware catches TimeLimitExceeded as a retryable exception (it is not a subclass of dramatiq.middleware.Interrupt which Retries skips) and re-queues the message. Without an idempotency key, the retry creates ch_B. Set time_limit conservatively — at least 3× the Stripe P99 latency plus your database write P99 — and rely on the content-hash idempotency key to close the ambiguity window rather than on timing alone.

2. pipeline() completion callbacks can re-dispatch a billing actor after the result is already written

Dramatiq's pipeline([fetch_customers.message(), charge_all_customers.message_with_options(pipe_ignore=False)]) passes the output of fetch_customers as the first argument to charge_all_customers. If a separate monitoring path also calls charge_all_customers.send(billing_period) directly — believing the pipeline stalled — two calls to charge_all_customers execute independently: one with the pipeline-injected argument and one with the manually provided argument. The Redis NX lock guards against this: the second actor to acquire the lock key exits immediately. Do not use pipeline() and direct .send() as independent trigger paths for the same billing actor — route all dispatch through a single entry point that acquires the lock.

3. Dramatiq's default message_id is a UUID; never use it as a Stripe idempotency key

Every .send() or .message() call generates a fresh UUID for the message's message_id. If you derive the Stripe idempotency key from message.message_id inside the actor — accessible via dramatiq.middleware.CurrentMessage.get_current_message().message_id — the key is different on the APScheduler-dispatched message and the manually dispatched message, even though both carry identical billing_period arguments. Both calls to Stripe then create distinct charges. The idempotency key must be derived from the stable billing parameters themselves — sha256(customer_id:amount_cents:billing_period:dramatiq-billing)[:32] — not from the message envelope.

4. group().get_results() timeout does not cancel in-flight actor executions

When you call group(...).run().get(block=True, timeout=120) from the dispatch process, a ResultTimeout exception is raised if the group does not complete within 120 seconds. However, all actor messages remain in the queue and workers continue processing them — the timeout only affects the calling process's wait, not the actor executions themselves. If the dispatch process then re-dispatches the group because get() timed out, original actors are still running and the Redis NX lock has already been acquired. The re-dispatch either fails to acquire the lock (correct behavior) or — if using group() directly without the lock — proceeds to dispatch a second round of charge actors. Always use the Redis NX lock pattern for billing groups; do not rely on get(block=True, timeout=N) as a dedup mechanism.

Enforcement tests

# tests/test_billing_actors.py
import hashlib
import pytest
from unittest.mock import patch, MagicMock, call
import fakeredis

def make_idempotency_key(customer_id: str, amount_cents: int, billing_period: str) -> str:
    return hashlib.sha256(
        f"{customer_id}:{amount_cents}:{billing_period}:dramatiq-billing".encode()
    ).hexdigest()[:32]

class TestDramatiqBillingIdempotency:
    def test_idempotency_key_stable_across_retries(self):
        """Same actor args on retry 1 and retry 2 produce the same idempotency key."""
        key1 = make_idempotency_key("cus_ABC", 2999, "2026-07")
        key2 = make_idempotency_key("cus_ABC", 2999, "2026-07")
        assert key1 == key2
        assert len(key1) == 32

    def test_idempotency_key_unique_per_customer(self):
        """Different customer IDs produce different idempotency keys."""
        key1 = make_idempotency_key("cus_ABC", 2999, "2026-07")
        key2 = make_idempotency_key("cus_XYZ", 2999, "2026-07")
        assert key1 != key2

    def test_card_declined_suppresses_exception(self):
        """CardError is caught and actor returns — Retries middleware does not re-enqueue."""
        import stripe
        with patch("stripe.Stripe") as mock_stripe_cls:
            mock_client = MagicMock()
            mock_stripe_cls.return_value = mock_client
            mock_client.charges.create.side_effect = stripe.error.CardError(
                "Your card was declined.", None, "card_declined"
            )
            with patch("actors.billing.db") as mock_db:
                mock_db.connect.return_value.__enter__ = MagicMock()
                mock_db.connect.return_value.__exit__ = MagicMock(return_value=False)
                # Actor should return None without raising — no retry triggered
                result = charge_customer("cus_FAIL", 2999, "2026-07", "vk_test")
            assert result is None
            assert mock_client.charges.create.call_count == 1

    def test_redis_nx_lock_prevents_double_dispatch(self):
        """Second actor.send() for the same period acquires no lock and returns early."""
        fake_redis = fakeredis.FakeRedis()
        with patch("actors.billing._redis", fake_redis):
            # First call acquires lock and proceeds
            with patch("actors.billing.db") as mock_db, \
                 patch("actors.billing.requests") as mock_requests, \
                 patch("dramatiq.group") as mock_group:
                mock_db.fetch_unpaid_customers.return_value = []
                charge_all_customers("2026-07")
                # Lock is now set
                assert fake_redis.exists("billing:lock:2026-07")

            # Second call for same period — lock already held — returns without dispatch
            with patch("dramatiq.group") as mock_group2:
                charge_all_customers("2026-07")
                mock_group2.assert_not_called()

    def test_vault_key_cap_includes_ten_percent_buffer(self):
        """Per-batch vault key cap is sum of all amounts × 1.10."""
        customers = [{"amount_cents": 2999} for _ in range(10)]
        total_cents = sum(c["amount_cents"] for c in customers)
        expected_cap = round((total_cents / 100) * 1.10, 2)
        assert expected_cap == pytest.approx(32.989)

FAQ

Can I use message.message_id as the Stripe idempotency key?

No. Dramatiq generates a fresh UUID for every .send() or .message() call. This means the APScheduler-dispatched message and the manually dispatched message for the same billing period have different message_id values — so using the message ID as the idempotency key produces distinct Stripe charges instead of deduplicating. Use a content-hash of the stable billing parameters: sha256(customer_id:amount_cents:billing_period:dramatiq-billing)[:32]. This key is identical across all dispatch paths and all retry attempts for the same billing intent.

Does max_retries=3 become safe once idempotency keys are in place?

Yes — for transient failures. All 3 retry attempts compute the same idempotency key from the same actor arguments, so Stripe returns ch_A on every retry for an APIConnectionError or network timeout. The remaining concern is permanent failures: card declines and invalid request errors should be caught inside the actor and handled by returning early rather than re-raising, so the Retries middleware does not schedule further attempts for a charge that will never succeed regardless of retry count.

How should I integrate APScheduler with Dramatiq safely?

Route all dispatch — scheduled and manual — through the single charge_all_customers.send(billing_period) call that acquires the Redis NX lock. APScheduler fires charge_all_customers.send("2026-07"); any manual trigger or monitoring webhook also calls charge_all_customers.send("2026-07"). The first actor to run acquires the lock and proceeds; all subsequent actors for the same period return immediately. Never bypass the lock by calling charge_customer.send() directly for individual customers in a period that has already been dispatched — the idempotency key at the customer level protects the Stripe call, but you still incur redundant Stripe round-trips and database load for every bypassed customer.

What happens if the vault key expires mid-group?

Actors that attempt stripe.charges.create() after the vault key expires receive a 401 Unauthorized from the proxy, which the Stripe Python SDK surfaces as stripe.error.AuthenticationError. The actor re-raises this exception (it is not a permanent billing error) and the Retries middleware re-enqueues it — but the key remains expired on subsequent attempts. Size the vault key TTL to cover your largest expected group duration with a 2× buffer: for 500 customers at 8 workers and 30ms Stripe latency, batch duration is roughly 2 seconds of Stripe time plus scheduling overhead and database writes. A 7200-second TTL covers batches up to tens of thousands of customers.

How do I bill the same customer legitimately twice in the same period?

Add a charge_reason discriminator to the idempotency key hash: sha256(f"{customer_id}:{amount_cents}:{billing_period}:{charge_reason}:dramatiq-billing"). Common values: subscription, overage, upgrade-proration. This produces a distinct idempotency key for each billing intent so the second charge is not deduplicated against the first. The Redis NX lock is keyed to billing:lock:{billing_period}:{charge_reason} for the same reason — a subscription run and an overage run for the same period need independent locks.

Does the vault key pattern work with dramatiq[redis] and dramatiq[rabbitmq] equally?

Yes. The vault key is an actor argument — a string passed through the message payload — so it works identically regardless of the broker backend. The Redis NX lock uses a separate Redis connection for the distributed lock; if your Dramatiq broker uses RabbitMQ, you still point the lock at Redis. The proxy itself is broker-agnostic: it intercepts the HTTPS call to proxy.keybrake.com/stripe/v1 regardless of how the actor was triggered or which broker delivered the message.