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

Luigi is Spotify's open-source Python workflow framework where work is modelled as tasks with dependency chains, output targets that define completion, and a central scheduler that manages execution state. When a billing task calls stripe.charges.create() inside run(), Luigi's target-based model introduces three distinct failure modes that no other framework replicates quite the same way.

This post covers three Luigi-specific failure modes — task retry re-execution, yield-based fan-out key sharing, and output target deletion triggering task re-runs — each of which causes 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-pipeline vault keys via a spend-cap proxy, without restructuring your task DAG.

Failure mode 1: Task retry re-invokes run() from line 1 after stripe.charges.create() has already succeeded

Luigi tasks support automatic retry via the retry_count parameter. When run() raises an unhandled Python exception, the Luigi central scheduler marks the task as FAILED and re-schedules it up to retry_count additional times. Each retry calls the full run() method from line 1 — there is no mid-function resume capability in Luigi.

The failure window is between stripe.charges.create() returning successfully and the task's output target being written. Consider a billing task that charges a customer and then writes the charge ID to a PostgreSQL table. If the database connection drops after the Stripe call succeeds — or if any subsequent operation in run() raises an exception — Luigi catches the error, waits the configured retry_delay, and re-executes run() on the next attempt. The Stripe call fires again. Without an idempotency key, Stripe creates ch_B. Both charge objects are valid; the customer has been charged twice; and the Luigi scheduler shows only one task with a retry count of 1.

# tasks/billing.py — UNSAFE: no idempotency key
import luigi
import stripe
import os
import psycopg2

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

class BillingTask(luigi.Task):
    customer_id = luigi.Parameter()
    amount_cents = luigi.IntParameter()
    billing_period = luigi.Parameter()
    retry_count = 3
    retry_delay = 60

    def output(self):
        return luigi.contrib.postgres.PostgresTarget(
            host=os.environ["DB_HOST"],
            database="billing",
            user=os.environ["DB_USER"],
            password=os.environ["DB_PASSWORD"],
            table="charges",
            update_id=f"{self.customer_id}:{self.billing_period}",
        )

    def run(self):
        # If this succeeds but the DB write below fails → run() retried → ch_B
        charge = stripe.charges.create(
            amount=self.amount_cents,
            currency="usd",
            customer=self.customer_id,
            description=f"Subscription {self.billing_period}",
        )
        # DB connection drop here → Luigi re-executes run() from line 1
        with psycopg2.connect(os.environ["DATABASE_URL"]) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "INSERT INTO charges VALUES (%s, %s, %s)",
                    [charge.id, self.customer_id, self.billing_period],
                )
            conn.commit()
        self.output().touch()

The fix is a content-hash idempotency key computed from the task parameters that are stable across all retries. Luigi passes the same parameter values on every retry — customer_id, amount_cents, and billing_period are immutable task parameters — so a key derived from these inputs is identical on every retry attempt. Stripe returns the original ch_A rather than creating ch_B. Permanent Stripe errors (card declined, invalid customer) should be caught and recorded to the output target rather than raised; a raised exception triggers another retry cycle that will re-attempt a charge that will never succeed.

# tasks/billing.py — SAFE: stable idempotency key + permanent-error handling
import hashlib
import luigi
import stripe
import requests
import os

class BillingTask(luigi.Task):
    customer_id = luigi.Parameter()
    amount_cents = luigi.IntParameter()
    billing_period = luigi.Parameter()
    vault_key = luigi.Parameter()   # issued once by upstream VaultKeyTask
    retry_count = 3
    retry_delay = 60

    def output(self):
        return luigi.contrib.postgres.PostgresTarget(
            host=os.environ["DB_HOST"],
            database="billing",
            user=os.environ["DB_USER"],
            password=os.environ["DB_PASSWORD"],
            table="charges",
            update_id=f"{self.customer_id}:{self.billing_period}",
        )

    def run(self):
        idempotency_key = hashlib.sha256(
            f"{self.customer_id}:{self.amount_cents}:{self.billing_period}:luigi-billing".encode()
        ).hexdigest()[:32]

        stripe_client = stripe.Stripe(
            api_key=self.vault_key,
            base_url="https://proxy.keybrake.com/stripe/",
        )

        try:
            charge = stripe_client.charges.create(
                amount=self.amount_cents,
                currency="usd",
                customer=self.customer_id,
                description=f"Subscription {self.billing_period}",
                idempotency_key=idempotency_key,
            )
            status, charge_id = "charged", charge.id
        except stripe.error.CardError as e:
            # Record the decline — do NOT raise; raising triggers another retry cycle
            status, charge_id = "card_declined", None
        except stripe.error.InvalidRequestError as e:
            status, charge_id = "invalid_request", None

        # Write output — if this fails, retry re-runs run() but idempotency key
        # absorbs the duplicate Stripe call and returns the original charge object
        with psycopg2.connect(os.environ["DATABASE_URL"]) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "INSERT INTO charges (charge_id, customer_id, billing_period, status) "
                    "VALUES (%s, %s, %s, %s) ON CONFLICT DO NOTHING",
                    [charge_id, self.customer_id, self.billing_period, status],
                )
            conn.commit()
        self.output().touch()

Failure mode 2: yield-based dynamic fan-out dispatches N concurrent BillingTask instances that all share one unrestricted STRIPE_SECRET_KEY

Luigi supports dynamic task dependencies via yield inside run(). When a parent BillingBatchTask yields a BillingTask instance for each customer, the Luigi scheduler receives all of them as pending requirements and starts executing them in parallel, subject only to the configured worker count. All parallel BillingTask instances read STRIPE_SECRET_KEY from the same process environment — there is no per-task key isolation.

The blast radius of a data error is the full fan-out width. If the amount_cents calculation in the upstream query has a unit bug — producing 299900 (dollars) instead of 2999 (cents) — all N parallel billing tasks each charge their customer the wrong amount before any single task has a chance to return an error. With a standard fan-out of 500 customers, 500 incorrect charges fire simultaneously. A Stripe-level spend cap cannot save you here if the key is unrestricted — there is no mechanism to halt the in-flight fan-out tasks once the error is detected in the first completed task.

# tasks/billing_batch.py — UNSAFE: fan-out shares one unrestricted key
import luigi
import psycopg2
import os
from tasks.billing import BillingTask

class BillingBatchTask(luigi.Task):
    billing_period = luigi.Parameter()

    def run(self):
        # Load all customers for this billing period
        with psycopg2.connect(os.environ["DATABASE_URL"]) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT customer_id, amount_cents FROM subscriptions "
                    "WHERE billing_period = %s AND status = 'active'",
                    [self.billing_period],
                )
                customers = cur.fetchall()

        # yield dispatches all N BillingTask instances to the scheduler in parallel
        # All share os.environ["STRIPE_SECRET_KEY"] — no per-task spend cap
        for customer_id, amount_cents in customers:
            yield BillingTask(
                customer_id=customer_id,
                amount_cents=amount_cents,
                billing_period=self.billing_period,
            )
        # A unit bug in amount_cents hits all 500 customers before any error surfaces

The fix is to issue a per-pipeline vault key with a cap equal to the total expected billing amount for the period, and pass it as a parameter to every BillingTask in the fan-out. The cap is calculated before the fan-out starts, based on the aggregate of all customer amounts for the billing period. Even if the amount calculation is correct, a per-pipeline cap means a runaway error can consume at most the budgeted amount before the proxy rejects further calls — without requiring any per-task key isolation in Luigi itself.

# tasks/vault_key.py — issue per-pipeline vault key before fan-out
import luigi
import requests
import os
import json

class VaultKeyTask(luigi.Task):
    billing_period = luigi.Parameter()
    max_spend_usd = luigi.FloatParameter()

    def output(self):
        return luigi.LocalTarget(f"/tmp/vault_key_{self.billing_period}.json")

    def run(self):
        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": self.max_spend_usd,
                "expires_in_seconds": 7200,
                "label": f"luigi-billing-{self.billing_period}",
            },
            timeout=10,
        )
        resp.raise_for_status()
        with self.output().open("w") as f:
            json.dump({"vault_key": resp.json()["vault_key"]}, f)

# tasks/billing_batch.py — SAFE: per-pipeline vault key passed to all fan-out tasks
import luigi
import psycopg2
import os
import json
from tasks.billing import BillingTask
from tasks.vault_key import VaultKeyTask

class BillingBatchTask(luigi.Task):
    billing_period = luigi.Parameter()

    def requires(self):
        # Compute max spend before issuing key — query the aggregate first
        with psycopg2.connect(os.environ["DATABASE_URL"]) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT SUM(amount_cents) / 100.0 * 1.10 FROM subscriptions "
                    "WHERE billing_period = %s AND status = 'active'",
                    [self.billing_period],
                )
                max_spend = cur.fetchone()[0] or 0.0
        return VaultKeyTask(
            billing_period=self.billing_period,
            max_spend_usd=max_spend,
        )

    def run(self):
        with self.input().open() as f:
            vault_key = json.load(f)["vault_key"]

        with psycopg2.connect(os.environ["DATABASE_URL"]) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT customer_id, amount_cents FROM subscriptions "
                    "WHERE billing_period = %s AND status = 'active'",
                    [self.billing_period],
                )
                customers = cur.fetchall()

        for customer_id, amount_cents in customers:
            yield BillingTask(
                customer_id=customer_id,
                amount_cents=amount_cents,
                billing_period=self.billing_period,
                vault_key=vault_key,
            )
        # All fan-out tasks share one vault key, but the proxy enforces the
        # daily_usd_cap across all calls made with that key

Failure mode 3: Output target deletion causes Luigi to re-run a billing task that already created a Stripe charge

Luigi's task completion model is entirely target-driven: a task is considered complete if and only if output().exists() returns True. Luigi stores no independent state about whether a task's run() method was successfully executed — only whether the output target currently exists. This design is deliberate and correct for idempotent data transformations, but it is dangerous for billing tasks where the external side effect (the Stripe charge) outlives the output target.

Targets are frequently deleted by legitimate operational actions: S3 lifecycle rules expire old billing records after 90 days; a DBA truncates a table partition to fix a data quality issue; an engineer runs a cleanup script that removes completed task markers; an S3 bucket is migrated and the old path becomes invalid. When the output target disappears, Luigi has no memory that run() was ever executed. On the next pipeline run for that billing period — triggered by a backfill, a manual re-run, or a regular schedule — Luigi sees a task with a missing output target, considers it incomplete, and calls run(). The Stripe charge fires again. ch_A from the original run and ch_B from the re-run both appear in your Stripe dashboard, both valid, often weeks or months apart.

# The re-run scenario — UNSAFE without idempotency key
#
# 1. June 1: BillingTask(customer_id="cus_ABC", billing_period="2026-06").run() executes
#    stripe.charges.create() → ch_A created, PostgresTarget written, task marked DONE
#
# 2. July 15: DBA truncates the charges table to rebuild from source → PostgresTarget gone
#    output().exists() now returns False for all June billing tasks
#
# 3. July 16: Engineer runs backfill for June period (audit or restatement)
#    Luigi sees BillingTask(customer_id="cus_ABC", billing_period="2026-06") as incomplete
#    run() executes again → stripe.charges.create() fires → ch_B created
#    Customer is charged twice, 45 days apart, both charges shown as "Succeeded"
#
# Without an idempotency key, Stripe treats the July request as a new charge
# because the idempotency window is only 24 hours from the original June 1 request
print("Target-based completion with no idempotency key = billing task can re-run any time")

Content-hash idempotency keys do not protect against this scenario because the Stripe idempotency key window is only 24 hours. A billing task re-run 45 days later will produce a key Stripe has never seen, and Stripe will create a new charge. The correct fix is a pre-flight check in run() that queries the billing database before calling Stripe — not relying on the target alone. If the database already has a row for this (customer_id, billing_period), the task returns early without charging, and recreates the output target.

# tasks/billing.py — SAFE: pre-flight DB check guards against target-deletion re-run
import hashlib
import luigi
import stripe
import psycopg2
import os

class BillingTask(luigi.Task):
    customer_id = luigi.Parameter()
    amount_cents = luigi.IntParameter()
    billing_period = luigi.Parameter()
    vault_key = luigi.Parameter()
    retry_count = 3
    retry_delay = 60

    def output(self):
        return luigi.contrib.postgres.PostgresTarget(
            host=os.environ["DB_HOST"],
            database="billing",
            user=os.environ["DB_USER"],
            password=os.environ["DB_PASSWORD"],
            table="charges",
            update_id=f"{self.customer_id}:{self.billing_period}",
        )

    def run(self):
        idempotency_key = hashlib.sha256(
            f"{self.customer_id}:{self.amount_cents}:{self.billing_period}:luigi-billing".encode()
        ).hexdigest()[:32]

        # Pre-flight check: did we already charge this customer for this period?
        with psycopg2.connect(os.environ["DATABASE_URL"]) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT charge_id, status FROM charges "
                    "WHERE customer_id = %s AND billing_period = %s",
                    [self.customer_id, self.billing_period],
                )
                existing = cur.fetchone()

        if existing:
            # Target was deleted but charge exists — recreate target, skip Stripe call
            self.output().touch()
            return

        stripe_client = stripe.Stripe(
            api_key=self.vault_key,
            base_url="https://proxy.keybrake.com/stripe/",
        )

        try:
            charge = stripe_client.charges.create(
                amount=self.amount_cents,
                currency="usd",
                customer=self.customer_id,
                description=f"Subscription {self.billing_period}",
                idempotency_key=idempotency_key,
            )
            status, charge_id = "charged", charge.id
        except stripe.error.CardError as e:
            status, charge_id = "card_declined", None
        except stripe.error.InvalidRequestError as e:
            status, charge_id = "invalid_request", None

        with psycopg2.connect(os.environ["DATABASE_URL"]) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "INSERT INTO charges (charge_id, customer_id, billing_period, status) "
                    "VALUES (%s, %s, %s, %s) ON CONFLICT DO NOTHING",
                    [charge_id, self.customer_id, self.billing_period, status],
                )
            conn.commit()
        self.output().touch()

Approach comparison

Gap analysis

1. luigi.build() called from scripts or notebooks bypasses the central scheduler's retry accounting

When luigi.build([BillingTask(...)]) is called from a Python script, Jupyter notebook, or cron job that does not connect to a running luigid daemon, Luigi runs in local mode with its own in-process worker. Local mode has no cross-run state sharing — if the script is interrupted and re-run, Luigi treats all tasks as fresh, including completed billing tasks whose output targets may have been partially written. The retry_count parameter is also handled differently: local mode may not apply the same retry backoff as the daemon. Always use the daemon-based scheduler in production billing pipelines, and ensure retry_count and retry_delay are explicitly set on all billing tasks rather than inherited from base class defaults.

2. LocalTarget and S3Target output targets can be lost silently without triggering task failure

When a billing task uses a LocalTarget (a file on disk) or an S3Target as its output, that target is invisible to your main application database. An engineer who clears the /tmp/ directory, a container restart that wipes ephemeral storage, or an S3 bucket migration can delete the target without any alarm or audit event. The task will silently re-execute on the next pipeline run. Use PostgresTarget or a custom database-backed target for billing tasks — write the target row to the same transactional database as the charge record in the same DB transaction, so the target cannot disappear without also losing the charge record that the pre-flight check queries.

3. CentralPlannerScheduler task history does not persist across luigid restarts

The Luigi central scheduler stores task state in memory, not on disk. When luigid restarts — after a deploy, a crash, or a container reschedule — all in-memory task history is lost. Tasks that were RUNNING at restart time are not automatically resumed or marked as failed; the workers lose their connection to the scheduler and the tasks become orphaned. If a billing task was mid-execution when luigid restarted, a new pipeline run will see the output target as absent (because run() did not complete far enough to write it) and re-execute the task. The pre-flight database check handles this correctly: it finds the existing charge record written before the restart and skips the Stripe call.

4. Luigi's Task.clone() creates a new task instance with overridden parameters that may break the idempotency key

Luigi's Task.clone(cls=OtherTask, **kwargs) method creates a new task instance, copying the current task's parameters and overriding with kwargs. It is commonly used in requires() to pass parameters down the dependency chain. If a developer clones a billing task with a modified billing_period or amount_cents — perhaps to fix a data quality issue — the cloned task will compute a different idempotency key. This is correct behavior if the intent is genuinely to charge for a different period or amount, but it creates a new charge even if the original was correct. Treat any modification to the parameters that feed the idempotency key derivation as equivalent to creating a new billing intent — validate the change against the existing charge record before allowing the cloned task to execute.

Enforcement tests

# test_luigi_billing.py
import hashlib
import pytest
from unittest.mock import patch, MagicMock

def make_idempotency_key(customer_id, amount_cents, billing_period):
    raw = f"{customer_id}:{amount_cents}:{billing_period}:luigi-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

def test_idempotency_key_stable_across_task_retries():
    key1 = make_idempotency_key("cus_ABC", 2999, "2026-07")
    key2 = make_idempotency_key("cus_ABC", 2999, "2026-07")
    assert key1 == key2, "Key must be identical across all run() re-executions"

def test_idempotency_key_unique_per_customer():
    key_abc = make_idempotency_key("cus_ABC", 2999, "2026-07")
    key_def = make_idempotency_key("cus_DEF", 2999, "2026-07")
    assert key_abc != key_def, "Different customers must produce different idempotency keys"

def test_card_declined_does_not_raise():
    from tasks.billing import BillingTask
    task = BillingTask(
        customer_id="cus_DECLINED",
        amount_cents=2999,
        billing_period="2026-07",
        vault_key="vk_test",
    )
    with patch("tasks.billing.stripe.Stripe") as mock_stripe, \
         patch("tasks.billing.psycopg2.connect") as mock_db:
        mock_stripe.return_value.charges.create.side_effect = (
            __import__("stripe").error.CardError("Declined", None, "card_declined", http_status=402)
        )
        mock_db.return_value.__enter__.return_value.cursor.return_value.__enter__.return_value.fetchone.return_value = None
        mock_db.return_value.__enter__.return_value.cursor.return_value.__enter__.return_value.execute.return_value = None
        # Should not raise — card decline must be caught and recorded
        try:
            task.run()
        except Exception:
            pytest.fail("BillingTask.run() must not raise on card decline")

def test_preflight_check_skips_stripe_call_on_existing_charge():
    from tasks.billing import BillingTask
    task = BillingTask(
        customer_id="cus_EXISTING",
        amount_cents=2999,
        billing_period="2026-06",
        vault_key="vk_test",
    )
    with patch("tasks.billing.stripe.Stripe") as mock_stripe, \
         patch("tasks.billing.psycopg2.connect") as mock_db:
        # DB returns an existing charge record
        mock_db.return_value.__enter__.return_value.cursor.return_value.__enter__.return_value.fetchone.return_value = ("ch_EXISTING", "charged")
        task.run()
        # Stripe must NOT be called if pre-flight check finds existing charge
        mock_stripe.return_value.charges.create.assert_not_called()

def test_vault_key_cap_covers_full_fan_out():
    from tasks.vault_key import VaultKeyTask
    import json
    task = VaultKeyTask(billing_period="2026-07", max_spend_usd=14995.0)
    with patch("tasks.vault_key.requests.post") as mock_post:
        mock_post.return_value.json.return_value = {"vault_key": "vk_test_cap"}
        mock_post.return_value.raise_for_status = lambda: None
        mock_out = MagicMock()
        mock_out.open.return_value.__enter__ = lambda s: MagicMock()
        mock_out.open.return_value.__exit__ = MagicMock(return_value=False)
        with patch.object(task, "output", return_value=mock_out):
            task.run()
        call_json = mock_post.call_args.kwargs["json"]
        assert call_json["allowed_endpoints"] == ["POST /v1/charges"]
        assert call_json["daily_usd_cap"] == 14995.0

Frequently asked questions

Is retry_count safe to use on billing tasks if I add idempotency keys?

Yes — with idempotency keys, task retries are safe for the Stripe call itself. When the retried run() fires stripe.charges.create() with the same idempotency key, Stripe returns the original charge object and no second charge is created. Retries can still cause issues in other parts of run(): database writes that are not idempotent (a plain INSERT without ON CONFLICT DO NOTHING) will raise a uniqueness violation on retry. Make every operation in a billing task's run() method idempotent — use ON CONFLICT DO NOTHING for database writes, idempotency keys for Stripe calls, and return early if the pre-flight check finds an existing charge.

Can I use Luigi's task_id as the Stripe idempotency key?

No. Luigi's task_id is a hash of the task class name and parameter values — it is stable across retries of the same task, which makes it superficially attractive as an idempotency key. However, it is not stable across re-runs triggered by target deletion: a re-run of a deleted task creates a new Task instance with the same task_id, but the Stripe idempotency window has expired if more than 24 hours have passed. Use a content-hash key derived from billing inputs (customer_id, amount_cents, billing_period) combined with a pre-flight database check for the long-window protection case. The task_id is useful for logging and correlation but must not be the sole Stripe idempotency key.

How do I handle a customer that should genuinely be charged twice in the same billing period?

Include a charge_reason or invoice_id parameter in the idempotency key derivation: SHA-256(customer_id:amount_cents:billing_period:charge_reason:luigi-billing)[:32]. A mid-month upgrade charge and an end-of-month subscription charge for the same customer in the same period produce different keys when the charge_reason differs. The pre-flight database check should also be scoped to (customer_id, billing_period, charge_reason) rather than just (customer_id, billing_period). Do not work around the idempotency key by adding a timestamp or random UUID — these make the key unstable across retries, defeating the purpose of idempotency entirely.

What happens if the VaultKeyTask output target is also deleted before the billing fan-out completes?

If the VaultKeyTask output target (the LocalTarget or S3Target containing the vault key JSON) is deleted while BillingBatchTask.run() is mid-execution, the remaining BillingTask instances in the fan-out will fail to read the vault key on the next retry cycle. The VaultKeyTask will re-execute and issue a new vault key from Keybrake. The new key has a fresh spend cap, so the proxy's accounting restarts. Add a pre-flight check in VaultKeyTask.run() that queries your database for an existing vault key for the billing period before issuing a new one — return the existing key if found, and issue a new one only if none exists. Store the issued vault key in the same transactional database as the charge records so it cannot be lost independently of the billing data.

Can I use Luigi's @requires decorator instead of defining requires() explicitly?

Yes, with one caveat for billing pipelines. The @luigi.util.requires(VaultKeyTask) decorator automatically wires the dependency and passes parameters from the parent task to VaultKeyTask. This works correctly when VaultKeyTask takes the same parameters as the task it decorates. The risk is that the max_spend_usd parameter must be pre-computed before BillingBatchTask is instantiated — the decorator cannot run dynamic queries at decoration time. Either pre-compute the cap outside Luigi and pass it as an explicit task parameter, or use an explicit requires() method that queries the database when constructing the VaultKeyTask dependency. Do not hardcode max_spend_usd as a constant — it must reflect the actual expected billing amount for the period to be effective as a circuit breaker.

Does Keybrake work if Luigi workers run in separate containers or machines?

Yes. Luigi's central scheduler dispatches tasks to worker processes that can run on any machine in the cluster. Each worker makes HTTP calls directly to proxy.keybrake.com using the vault key passed as a task parameter. The proxy is a network service, not a local library, so it works regardless of how Luigi workers are distributed. The per-pipeline vault key cap is enforced at the proxy level across all worker calls combined — if five workers each try to charge max_spend_usd / 5, the proxy correctly aggregates all five and rejects any call that would exceed the total cap. The label on the vault key (luigi-billing-2026-07) makes it easy to identify in the Keybrake audit log which pipeline run issued which calls.