I’m developing and maintaining a Python package to easily integrate your Django Application with Google Cloud Task.
Some features:
- Easily push tasks to Cloud Task using a decorator
- Automatically route all tasks from a single endpoint
- Ease scheduling with native python datetime
- Named task to avoid duplicated
- Local development support with Redis Queue
Simple and beautiful task definition
from datetime import timedelta
from django.utils.timezone import now
from cloudtask import (
CloudTaskRequest,
task)
@task(queue='default')
def add(request: CloudTaskRequest, a: int = 5, b: int = 4) -> None:
print(f'Running task with args {a=} and {b=}')
print(a + b)
# pushing task to queue
add(a=2, b=4).delay()
# executing the task immediately without push to queue
add(a=30)()
# scheduling the task
at = now() + timedelta(days=1)
add(b=15).schedule(at=at)
See the repository and full documentation on my Github page.
Thanks to the project bazaar.
If it was helpful or if you found it interesting, also leave an 🙋🏽♂️❤️ in the comments. Thank you, looking forward to the next article. Enjoy and follow my work.
Thanks,
Stay safe!
Top comments (0)