celery-singleton

celery-singleton is a base class for Celery tasks that ensures only one instance of a given task (matched by task name and arguments) can be queued or running at any moment. Instead of the caller having to build their own deduplication logic, tasks that subclass Singleton transparently return the AsyncResult of an already in-flight duplicate rather than spawning a second one.

Under the hood it uses Redis-backed distributed locking: each call to delay()/apply_async() is hashed into a lock key from the task name plus its JSON-serialized arguments, acquired atomically with SETNX, and released automatically when the task finishes (success or failure) or after an optional expiry. The storage layer is pluggable via an abstract BaseBackend, so teams can swap in a non-Redis datastore if needed.

What You Get

• A drop-in Singleton task base class for Celery (base=Singleton on any @app.task)
• Automatic Redis-based distributed locking keyed by task name + arguments
• A clear_locks() helper for clearing stale locks on worker startup
• A pluggable BaseBackend interface for swapping in non-Redis storage

Common Use Cases

• Preventing duplicate periodic/cron task executions
• Deduplicating tasks triggered by unreliable or retrying producers
• Enforcing one-job-per-resource background processing
• Recovering gracefully from crashed workers via lock expiry

Under The Hood

Architecture The library is a thin single-purpose extension of Celery’s Task base class. Singleton (celery_singleton/singleton.py) overrides apply_async() to compute a deterministic lock key via util.generate_lock() (an MD5 hash of task name plus JSON-serialized args/kwargs), attempt to acquire it through a pluggable singleton_backend (get_backend() in backends/__init__.py, cached as a module-level singleton), and either delegate to the base Task.apply_async() or return the AsyncResult for the task that already holds the lock; on_success/on_failure release the lock via release_lock(). Configuration (Config in config.py) reads all tunables lazily off app.conf, and storage is abstracted behind BaseBackend (an ABC with lock/unlock/get/clear), with RedisBackend as the only concrete implementation shipped. The separation of task logic, config, and storage is clean, though the module-level backend cache would conflict if an app ever needed two differently-configured backends in one process.

Tech Stack Pure Python (Poetry-managed) targeting Python ^3.6, with celery>=4 and an unpinned redis client as its only runtime dependencies. There’s no web framework, ORM, or CLI — it’s a library extension point rather than a standalone app. Locking uses raw redis-py SET NX EX semantics, json.dumps(sort_keys=True) plus hashlib.md5 for deterministic lock-key hashing, and Kombu’s uuid helper for task ID generation. CI (.travis.yml) runs a matrix across Python 3.6-3.8 and Celery 4/5 against a live Redis service, using Poetry for dependency management and pytest as the runner.

Code Quality The test suite (695 lines across test_singleton.py, test_backends.py, test_config.py, and a shared conftest.py) favors integration-style tests against a real Celery Task and a live Redis fixture that flushes between tests, supplemented by unittest.mock for edge cases — a reasonably thorough approach for a locking library. There are no type hints anywhere in the source and no linter or type checker wired into CI. Naming is mostly consistent and descriptive, aside from a persistent “aquire” typo across method names. Error handling is explicit: lock_and_run() catches exceptions during apply_async to release the lock before re-raising, and DuplicateTaskError carries the conflicting task ID.

API Design The public API is minimal and idiomatic — you opt in by passing base=Singleton to the standard @app.task decorator, the exact pattern Celery users already know, with just three optional class attributes (unique_on, raise_on_duplicate, lock_expiry) to learn. Scoping uniqueness via unique_on binds task arguments through inspect.signature to select only named parameters, avoiding any need to restructure task signatures. The main rough edge is that a deduplicated call silently returns someone else’s AsyncResult unless raise_on_duplicate is explicitly enabled, and there are no type stubs to guide IDE autocomplete.