Celery Stubs

0.1.3 · maintenance · verified Wed Apr 15

celery-stubs provides type hints for the popular Celery distributed task queue, allowing static type checkers like MyPy to validate Celery-related code. This package is currently at version 0.1.3, released in February 2023, and exhibits a slow, infrequent release cadence.

Warnings

gotcha This package provides *only* type stubs (`.pyi` files) for static analysis and does not add any runtime functionality or client-side proxy behavior to Celery. It is purely for enhancing type checking capabilities.
Fix: Understand that `celery-stubs` is a development dependency for type checking, not a runtime dependency.
gotcha The `celery-stubs` package is currently in '3 - Alpha' development status. This indicates that its API might not be stable, and future minor versions could introduce breaking changes.
Fix: Pin the version of `celery-stubs` in your project's dependencies to prevent unexpected breaking changes, e.g., `celery-stubs==0.1.3`.
gotcha For more comprehensive and actively maintained type stubs covering the broader Celery ecosystem (including `amqp`, `kombu`, `billiard`, etc.), consider using the `celery-types` package by `sbdchd` instead. It appears to be more actively developed and widely used.
Fix: Evaluate `celery-types` (`pip install celery-types`) as a potential alternative if `celery-stubs` does not meet your type-hinting needs for the full Celery ecosystem.

Install

pip install celery-stubs Install latest version

Imports

Celery
```
from celery import Celery
```
celery-stubs provides type hints for the 'celery' library itself. You import from 'celery', and the type checker uses the stubs.
Task
```
from celery import Task
```
celery-stubs provides type hints for the 'celery' library itself. You import from 'celery', and the type checker uses the stubs.

Quickstart

This quickstart demonstrates a basic Celery application using type hints. With `celery-stubs` installed, a static type checker like MyPy can validate the type annotations within your Celery tasks and interactions with Celery objects, helping catch type-related errors before runtime.

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

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

@app.task
def add(x: int, y: int) -> int:
    return x + y

@app.task
def process_data(data: Dict[str, Any]) -> None:
    print(f"Processing: {data['value']}")

# To demonstrate type checking, you would typically run mypy:
# mypy your_script_name.py
# If celery-stubs is installed, mypy will use it to validate types
# related to Celery objects (e.g., app.task decorator, AsyncResult, etc.)

# Example of type-hinted usage (not directly runnable without a Celery worker):
if __name__ == "__main__":
    result = add.delay(1, 2)
    print(f"Task ID: {result.id}")
    # result_value: int = result.get() # This line would be type-checked by mypy

    # Example of incorrect usage that mypy would catch with stubs:
    # bad_result = add.delay('a', 'b') # mypy would flag this as type error

view raw JSON →