celery-singleton

library0.3.1pypypi✓ verified 86d ago

Celery Singleton is a Python library that provides a base class for Celery tasks, ensuring that only one instance of a specific task can be queued or running at any given time. It achieves this by using Redis for distributed locking, leveraging the task's name and arguments to determine uniqueness. The current version is 0.3.1, released in January 2021, and its development appears to be in maintenance with recent activity on its GitHub issues.

database workflow aws

pip install celery-singleton

github ↗package ↗

INSTALL

IMPORT

SIG · CELERY-SINGLETON

celery-singleton

databasepythonv0.3.1

Install

3.9s avg

Import

737ms

Disk

42MB

Pass rate

10/ 10

Env Coverage10 / 10

glibc

3.9–3.13

musl

3.9–3.13

Install & Compatibility

Where this runs

tested against v0.3.1 · pip install

no network on importno background threads

Install × environment matrix

Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux

musl

py 3.10–3.920 runs

installs and imports cleanly · install 0.0s · import 0.762s · 45.5MB

glibc

py 3.10–3.920 runs

installs and imports cleanly · install 3.9s · import 0.711s · 46MB

42MB installed

● package 42MB

Code

Verified usage

Verified import paths — ran on the pinned version, not inferred.

Singleton

✓ from celery_singleton import Singleton

DuplicateTaskError

✓ from celery_singleton import Singleton, DuplicateTaskError

Import DuplicateTaskError only if you set `raise_on_duplicate=True` on your task.

This quickstart demonstrates how to define a Celery task as a singleton using `celery-singleton`. It shows that subsequent calls to `delay()` with identical arguments will return the `AsyncResult` of the already queued or running task, rather than queuing a new one. Ensure your Celery app is configured with a Redis broker and result backend.

import time from celery_singleton import Singleton from celery import Celery import os # Assuming a local Redis for Celery broker and result backend celery_app = Celery( 'my_app', broker=os.environ.get('CELERY_BROKER_URL', 'redis://localhost:6379/0'), backend=os.environ.get('CELERY_RESULT_BACKEND', 'redis://localhost:6379/1') ) @celery_app.task(base=Singleton) def do_stuff(*args, **kwargs): time.sleep(4) return 'I just woke up' if __name__ == '__main__': # Example of running tasks print('Calling do_stuff(1, 2, 3, a="b") the first time...') async_result = do_stuff.delay(1, 2, 3, a='b') print(f'First call result ID: {async_result.id}') print('Calling do_stuff(1, 2, 3, a="b") the second time (should return existing task)...') async_result2 = do_stuff.delay(1, 2, 3, a='b') print(f'Second call result ID: {async_result2.id}') assert async_result.id == async_result2.id print('Assertion successful: Duplicate task call returned the same AsyncResult ID.') print(f'Task 1 status: {async_result.status}') print(f'Waiting for task 1 to complete...') print(f'Task 1 result: {async_result.get()}')

Debug

Known issues

gotchaAll arguments passed to a singleton task must be JSON serializable, as celery-singleton uses their JSON representation to generate a unique lock key in Redis. Non-serializable arguments will cause task queuing to fail.

fix

Ensure all `*args` and `**kwargs` for tasks based on `Singleton` are JSON serializable. For complex objects, pass their IDs and re-fetch them within the task.

affects: 0.3.1

gotchaCelery Singleton relies on Redis for distributed locking. If your Celery app does not use Redis as its result backend or broker, you must explicitly configure a Redis URL for `celery-singleton` using the `singleton_backend_url` setting in your Celery config.

fix

Configure `app.conf.singleton_backend_url = 'redis://your_redis_host:port/db'` in your Celery application settings if Redis is not your default broker/backend.

affects: 0.3.1

gotchaIf a Celery worker is forcefully terminated (e.g., via a hard time limit or unexpected crash) while executing a singleton task, the Redis lock for that task might not be released, leading to a 'deadlock' where subsequent identical tasks are permanently blocked.

fix

Always set a `lock_expiry` on your singleton tasks, e.g., `@celery_app.task(base=Singleton, lock_expiry=300)`. The `lock_expiry` should be slightly longer than your task's expected maximum runtime.

affects: 0.3.1

Errors

Common errors & fixes

TypeError: Object of type <YourObject> is not JSON serializable

You are passing a Python object as a task argument that cannot be converted to JSON. celery-singleton needs JSON serializable arguments to create a consistent lock key.

fix

Refactor your task to accept only JSON-serializable data (e.g., IDs, strings, numbers, lists, dictionaries). If you need to work with complex objects, pass their unique identifiers and load the full object within the task body.

celery.exceptions.DuplicateTaskError: Task <task_name> with args <args> and kwargs <kwargs> is already running or queued.

This error occurs when you attempt to call a singleton task that is already running or queued, and the task is configured with `raise_on_duplicate=True`.

fix

If you want to handle duplicates by receiving the `AsyncResult` of the existing task instead of an error, remove `raise_on_duplicate=True` from your task decorator. If you explicitly want to prevent queuing and catch the error, ensure your calling code handles `DuplicateTaskError`.

Task always remains in 'PENDING' state and never executes, even when no other instance is running.

This often indicates that the Redis lock for the task was not released due to a previous worker crash or hard termination, leading to a stale lock. Another cause can be a misconfigured `singleton_backend_url` preventing `celery-singleton` from finding/creating locks.

fix

Ensure all singleton tasks have a `lock_expiry` set to prevent permanent stale locks. Check your Celery and `celery-singleton` Redis backend configurations. Manually clear stale locks from your Redis instance if necessary, using `DEL <lock_key>` where `<lock_key>` typically includes the task name and arguments hash.

Upgrade

Version history

0.3.1latest on PyPI · released Jan 14, 2021

Audit

Dependencies

celeryrequiredCore task queue functionality.

redisrequiredUsed for distributed locking to ensure task uniqueness.

Agent activity

24 hits · last 30 days

node

OpenAI (training)

Resources

githubhttps://github.com/steinitzu/celery-singleton ↗

packagehttps://pypi.org/project/celery-singleton/ ↗