celery_stubs_typed_task_definition_with_mypy_validation.py

1from celery import Celery, Task
2
3app = Celery("tasks", broker="pyamqp://guest@localhost//")
4
5@app.task
6def add(x: int, y: int) -> int:
7    return x + y
8
9# With celery-stubs installed, mypy can verify the following:
10# 1. 'delay' is recognized as a method on the task
11# 2. Argument types (int, int) are validated
12result = add.delay(4, 4)
13
14# 3. The return type of 'get()' is known to be an int
15value: int = result.get(timeout=1)
16print(f"The sum is {value}")