Skip to content

Instantly share code, notes, and snippets.

@vartec
Last active July 4, 2024 00:07
Show Gist options
  • Save vartec/0b93c6e202ec0a8c6aca04bc3eeba16a to your computer and use it in GitHub Desktop.
Save vartec/0b93c6e202ec0a8c6aca04bc3eeba16a to your computer and use it in GitHub Desktop.
Celery .delay() vs .apply_async() re MyPy
from celery import Celery
from typing import Any
app = Celery('tasks', broker='pyamqp://guest@localhost//')
@app.task
def add(x: int, y: int) -> int:
return x + y
# Usage examples
result = add.delay(1, 2) # This will pass mypy validation
result = add.apply_async(args=(1, 2)) # This will also pass
# These will raise mypy errors
result = add.delay("1", 2) # Error: Argument 1 to "delay" of "TaskType" has incompatible type "str"; expected "int"
result = add.delay(1, "2") # Error: Argument 2 to "delay" of "TaskType" has incompatible type "str"; expected "int"
result = add.apply_async(args=(1, "2")) # Error: Argument "args" to "apply_async" of "TaskType" has incompatible type "Tuple[int, str]"; expected "Tuple[int, int]"
# This will pass mypy but fail at runtime
result = add.apply_async(kwargs={"x": 1, "y": 2, "z": 3}) # mypy doesn't check kwargs keys
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment