celery_types_quickstart_typed_tasks_and_async_results.py

1from celery import Celery
2from celery.canvas import Signature
3
4app = Celery("tasks", broker="pyamqp://guest@localhost//")
5
6@app.task
7def add(x: int, y: int) -> int:
8    return x + y
9
10# Example of using the types with a task call
11result = add.delay(4, 4)
12# Type checkers can now infer that 'result' is an AsyncResult and 
13# 'result.get()' will return an int.
14val: int = result.get(timeout=1)
15
16# Example of using signatures with type hints
17sig: Signature = add.s(2, 2)
18sig_result: int = sig.delay().get()
19
20print(f"Result 1: {val}")
21print(f"Result 2: {sig_result}")