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)

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.

• See Here, all my resources.
• Learn With Me on Youtube.
• Follow Me on Instagram to stay updated.

Thanks,
Stay safe!