DEV Community

AM
AM

Posted on • Updated on

Django Cloud Task Queue

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)
Enter fullscreen mode Exit fullscreen mode

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!

Discussion (0)