Effective Celery Task Throttling: Parameter-Based Rate Limiting


Celery is a powerful, distributed task queue system that enables asynchronous task processing. However, managing the rate at which tasks are executed, especially those with the same parameters, can be challenging. This guide explores a solution for parameter-based rate limiting in Celery, ensuring your tasks respect external rate limits and avoid overwhelming systems.

Understanding the Problem

When working with Celery, you might encounter scenarios where tasks need to be throttled based on specific parameters. For example, an external API might limit the number of requests per minute per user. Without proper throttling, your Celery tasks could exceed these limits, leading to errors and potential bans.

The Solution: Parameter-Based Rate Limiting

Our solution involves implementing a decorator that enforces rate limits on Celery tasks based on specified parameters. This ensures that tasks with the same parameters do not exceed the allowed rate.

Step-by-Step Breakdown

1. Parsing the Rate

The parse_rate function converts a rate string (e.g., “5/m”, “10/2h”) into a tuple representing the allowed number of requests and the time period in seconds.

def parse_rate(rate: str) -> Tuple[int, int]:
    num, period = rate.split("/")
    num_requests = int(num)
    if len(period) > 1:
        duration_multiplier = int(period[:-1])
        duration_unit = period[-1]
        duration_multiplier = 1
        duration_unit = period[-1]
    duration_base = {"s": 1, "m": 60, "h": 3600, "d": 86400}[duration_unit]
    duration = duration_base * duration_multiplier
    return num_requests, duration


The rate string is split into two parts: the number of requests (num) and the time period (period).
The period is converted to seconds using a dictionary that maps time units to their equivalent in seconds.

2. Throttling Decorator

The throttle_task decorator applies the throttling logic to a task function.

def throttle_task(rate: str, key: Any = None) -> Callable:
    def decorator_func(func: Callable) -> Callable:
        def wrapper(*args, **kwargs) -> Any:
            sig = inspect.signature(func)
            bound_args = sig.bind(*args, **kwargs)
            task = bound_args.arguments["self"]
            key_value = None
            if key:
                    key_value = bound_args.arguments[key]
                except KeyError:
                    raise KeyError(f"Unknown parameter '{key}' in throttle_task decorator of function {task.name}. `key` parameter must match a parameter name from function signature: '{sig}'")
            delay = get_task_wait(task, rate, key=key_value)
            if delay > 0:
                task.request.retries = task.request.retries - 1
                logger.info("Throttling task %s (%s) via decorator for %ss", task.name, task.request.id, delay)
                return task.retry(countdown=delay)
                return func(*args, **kwargs)
        return wrapper
    return decorator_func


  • The decorator wraps the original task function.
  • It inspects the function’s parameters to bind the arguments and retrieve the task instance.
  • It calculates the delay required based on the rate and key (if provided).
  • If the task needs to be throttled, it logs the throttling event and reschedules the task after a delay. Otherwise, it proceeds with the task execution.

3. Rate Checking

The is_rate_okay function checks if the task can proceed based on the current rate limit.

def is_rate_okay(task: Task, rate: str = "1/s", key=None) -> bool:
    key = f"celery_throttle:{task.name}{':' + str(key) if key else ''}"
    r = make_redis_interface("CACHE")
    num_tasks, duration = parse_rate(rate)
    count = r.get(key)
    if count is None:
        r.set(key, 1)
        r.expire(key, duration)
        return True
        if int(count) <= num_tasks:
            r.incr(key, 1)
            return True
            return False


  • Constructs a Redis key based on the task name and optional key.
  • Retrieves the current count of tasks from Redis.
  • If the count is below the threshold, increments the count and allows the task to proceed. Otherwise, throttles the task.

4. Task Rescheduling

The set_for_next_window function ensures tasks are rescheduled for the next time window.

def set_for_next_window(r: Redis, throttle_key: str, schedule_key: str, n: datetime) -> float:
    ttl = r.ttl(throttle_key)
    if ttl < 0:
        return 0
    r.set(schedule_key, str(n + timedelta(seconds=ttl)))
    return ttl


Calculates the time-to-live (TTL) for the current throttle key.
Sets a schedule for the next window if TTL is valid. If not, runs the task immediately.

How to Use the Throttling Solution

To use this parameter-based rate limiting solution in your Celery tasks, follow these steps:

1. Define Your Celery Task:

Define the task you want to throttle.

from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379/0')

@throttle_task(rate='5/m', key='user_id')
def my_task(self, user_id, data):
    # Task logic here
  • The @throttle_task decorator is applied to the Celery task.
  • The rate parameter specifies the rate limit (e.g., 5 requests per minute).
  • The key parameter indicates the argument (e.g., user_id) used to differentiate between tasks.

2. Run Your Celery Worker:

Start the Celery worker to process tasks.

celery -A tasks worker --loglevel=info

3. Queue Tasks:

Queue tasks with different parameters to see the throttling in action.

my_task.apply_async(args=[1, 'some data'])
my_task.apply_async(args=[1, 'more data'])
my_task.apply_async(args=[2, 'other data'])
  • Tasks with the same user_id will be throttled according to the specified rate limit.
  • Tasks with different user_id values will be processed independently.


Implementing parameter-based rate limiting in Celery ensures your tasks respect external rate limits and avoid overwhelming systems. By understanding and applying this solution, you can maintain efficient and compliant task processing in your applications.

Explore More Celery Posts

Advanced Celery Task Throttling with Multiple Parameters

Explore how to implement advanced Celery task throttling using multiple parameters to ensure efficient and compliant task processing.

Read More
Celery: Efficient Task Queue Management with Python

Learn how to use Celery for task queue management in Python applications. Discover the best practices, and implementation with examples. Improve proc…

Read More
Celery with Redis for Efficient Task Queue Management in Python

Learn how to use Celery with Redis for efficient task queue management and how to monitor task results and failures in a Celery application.

Read More
Optimize Your Celery Setup: Tips & Techniques

Maximize task efficiency and minimize failure with these Celery best practices and third-party tools. Implement Redis, retries and callbacks with exa…

Read More