Type stubs for Celery and its related packages

0.26.0 · active · verified Fri Apr 10

Celery-types provides PEP 561 compliant type stubs for Celery and its core dependencies, including `kombu`, `amqp`, `billiard`, `vine`, and `django-celery-results`. This library enhances type checking for Celery applications, allowing tools like MyPy to verify correct usage of Celery APIs. It is actively maintained with regular updates to support new Python and Celery versions.

Warnings

gotcha For full generic type support in certain Celery classes (e.g., `Celery`, `Task`, `AsyncResult`), `celery-types` suggests a runtime monkey patch for `__class_getitem__`. Without this, generic type hints on these classes might not be fully recognized by type checkers.
Fix: Refer to the `celery-types` GitHub README for the latest monkey-patching snippet to enable full generic support for relevant Celery classes in your application's entry point. This typically involves iterating over a list of Celery classes and setting their `__class_getitem__` attribute.
gotcha celery-types provides *type stubs*, not runtime code. You should import Celery symbols from their original packages (e.g., `from celery import Celery`), and the installed `celery-types` package will automatically provide the type information to your type checker (e.g., MyPy). Attempting to import directly from `celery_types` will likely result in an `ImportError` or incorrect behavior.
Fix: Always import Celery classes and functions directly from the `celery` (or `kombu`, `django_celery_results`, etc.) packages. `celery-types` is a dependency for your type checker, not for runtime execution.
breaking celery-types is designed to provide type hints for specific versions of the underlying Celery library. Celery itself has undergone significant breaking changes, especially regarding Python version support (e.g., Celery 5.0 dropped Python 2.7, Celery 5.6.0 requires Python 3.9+). Using `celery-types` with an incompatible major version of `celery` can lead to incorrect or missing type information.
Fix: Always ensure your `celery-types` version is compatible with your installed `celery` version. Consult the `celery-types` release notes or `pyproject.toml` for supported `celery` ranges. Upgrade both `celery` and `celery-types` in tandem, following `celery`'s migration guides for runtime breaking changes.
gotcha While `celery-types` helps with type checking, it does not resolve runtime serialization issues in Celery. If you pass complex Python objects as task arguments, default JSON serialization might fail. Celery's default serializer changed from `pickle` to `json` in version 4.0. Using `pickle` is possible but less secure.
Fix: For complex objects, ensure they are JSON-serializable, or explicitly configure Celery to use a different serializer (e.g., `pickle` if security is less of a concern for internal communication), or pass only serializable data like IDs and retrieve complex objects within the task itself.

Install

pip install celery-types Install stable version

Imports

Celery
```
from celery import Celery
```
celery-types provides type annotations for the Celery library; you import from 'celery' itself, not 'celery_types'.
Task
```
from celery import Task
```
celery-types provides type annotations for the Celery library; you import from 'celery' itself, not 'celery_types'.
AsyncResult
```
from celery.result import AsyncResult
```
celery-types provides type annotations for the Celery library; you import from 'celery' itself, not 'celery_types'.

Quickstart

This quickstart demonstrates a basic Celery application with a type-hinted task. With `celery-types` installed, your type checker will validate the arguments and return types of Celery tasks and related objects like `AsyncResult`.

from celery import Celery, Task
from typing import Any, Dict

# Configure Celery app (replace with your broker/backend)
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/0')
)

# Define a type-hinted task
@app.task
def add(x: int, y: int) -> int:
    return x + y

# Example of calling the task with type hints
if __name__ == '__main__':
    # The type checker (e.g., MyPy) will use celery-types to validate arguments
    result: AsyncResult[int] = add.delay(10, 20)
    print(f"Task ID: {result.id}")
    print(f"Result: {result.get()}")

    # Incorrect usage (type checker would flag this if celery-types is installed)
    # add.delay(10, "20") # Mypy would warn about this

view raw JSON →