Celery 5 on Redis with fakeredis-backed idempotency, tested offline in eager mode: working versions (tested 2026-10-15)

celery==5.6.3
redis==8.1.0
fakeredis==2.40.0
pytest==9.1.1

import os
from celery import Celery

REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379/0")

app = Celery("app", broker=REDIS_URL, backend=REDIS_URL)
app.conf.update(
    task_acks_late=True,           # redeliver if a worker dies mid-task
    worker_prefetch_multiplier=1,  # don't hoard long tasks on one worker
    result_expires=3600,
)

import os
import redis

class IdempotencyLock:
    def __init__(self, client: redis.Redis, ttl: int = 3600, prefix: str = "idem:"):
        self.client = client
        self.ttl = ttl
        self.prefix = prefix

    def claim(self, key: str) -> bool:
        # SET NX returns True on success and None when the key already exists
        return bool(self.client.set(self.prefix + key, "1", nx=True, ex=self.ttl))

    def release(self, key: str) -> None:
        self.client.delete(self.prefix + key)

_default: IdempotencyLock | None = None

def get_lock() -> IdempotencyLock:
    global _default
    if _default is None:
        url = os.environ.get("REDIS_URL", "redis://localhost:6379/1")
        _default = IdempotencyLock(redis.Redis.from_url(url))
    return _default

from celery_app import app
from dedupe import get_lock

def charge_gateway(order_id: str, cents: int) -> None:
    raise NotImplementedError("wire your payment provider here")

@app.task
def record_payment(order_id: str, cents: int) -> str:
    lock = get_lock()
    key = f"payment:{order_id}"
    if not lock.claim(key):
        return "duplicate"
    try:
        charge_gateway(order_id, cents)
    except Exception:
        lock.release(key)  # let a retry claim the key again
        raise
    return "charged"

import fakeredis, pytest
import tasks
from celery_app import app
from dedupe import IdempotencyLock

@pytest.fixture(autouse=True)
def eager():
    app.conf.update(task_always_eager=True, task_eager_propagates=True)
    yield
    app.conf.update(task_always_eager=False)

@pytest.fixture
def fake_redis(monkeypatch):
    fake = fakeredis.FakeRedis()
    fake.flushall()  # fakeredis instances may share data, so start clean
    monkeypatch.setattr(tasks, "get_lock", lambda: IdempotencyLock(fake))
    return fake

@pytest.fixture(autouse=True)
def gateway(monkeypatch):
    calls = []
    monkeypatch.setattr(tasks, "charge_gateway", lambda order_id, cents: calls.append(order_id))
    return calls

def test_runs_inline_in_eager_mode(fake_redis, gateway):
    assert tasks.record_payment.delay("o1", 500).get() == "charged"
    assert gateway == ["o1"]

def test_duplicate_is_skipped(fake_redis, gateway):
    # SET NX returns None (not False) on conflict, so the helper must coerce to bool
    assert tasks.record_payment.delay("o2", 500).get() == "charged"
    assert tasks.record_payment.delay("o2", 500).get() == "duplicate"
    assert gateway == ["o2"]

def test_lock_has_ttl(fake_redis, gateway):
    # without a TTL, a crashed worker would block this order forever
    tasks.record_payment.delay("o4", 500).get()
    assert 0 < fake_redis.ttl("idem:payment:o4") <= 3600

def test_failed_run_releases_claim(monkeypatch, fake_redis):
    calls = []
    def flaky(order_id, cents):
        calls.append(order_id)
        if len(calls) == 1:
            raise RuntimeError("gateway down")
    monkeypatch.setattr(tasks, "charge_gateway", flaky)
    with pytest.raises(RuntimeError):
        tasks.record_payment.delay("o3", 100).get()
    assert tasks.record_payment.delay("o3", 100).get() == "charged"
    assert len(calls) == 2

[pytest]
pythonpath = .