Celery backend for backgrounds tasks

[alexalligator]

Hi there!

I'm using celery for background tasks on my Django-Wagtail site and am wondering if anyone has any tips on how to make a JobBackend for celery.

What I'm stuck on is how to take a callable (that has not been decorated with celery's @task decorator) and then applying it with .delay().

I can't get my head around how you would go about registering and firing an arbitrary callable as a task. I've had a look at the DjangoRQJobBackend source code and see that there you can just tell your queue to enqueue the callable with its args and kwargs. Is anyone aware of a similar feature in celery?

Anyone got any tips?

Would massively appreciate any help on this. 🙏

from wagtail_localize.tasks import BaseJobBacked


class MyJobBackend(BaseJobBackend):
    def __init__(self, options):
        # Any set up code goes here. Note that the 'options' parameter contains the value of WAGTAILLOCALIZE_JOBS["OPTIONS"]
        pass

    def enqueue(self, func, args, kwargs):
        # func is a function object to call
        # args is a list of positional arguments to pass into the function when it's called
        # kwargs is is a dict of keyword arguments to pass into the function when it's called
        pass

[Nigel2392]

I've figured it out! It might break anytime though - documentation on this library is relatively sparse and I don't see any compatibility promise.

The thing is, I also got stuck on using the callable. Forget about it! We are being passed an already defined function - let's just use that instead. This allows us to easily create a wrapper for this object.

Anyways - it's a relatively simple bit of code. It's also using pickle to serialize an object - security might be implicated.
It also only works for one of the use cases - when directly translating the tree.
It does not work when it synchronises two page trees. For this - you'd probably have to check the length of args and then call the appropriate function

from celery import shared_task
from django.contrib.auth import get_user_model
from wagtail_localize.tasks import BaseJobBackend

import pickle # :(


class CeleryBackend(BaseJobBackend):

    def enqueue(self, func, args, kwargs):
        if len(args) == 4:
            page_id, locales, components, user = args
            locale_ids = [l.pk for l in locales]
            components_data = pickle.dumps(components)
            user_id = user.pk
    
            do_translate_page_subtree.delay(page_id, locale_ids, components_data, user_id)
        func(*args, **kwargs)


@shared_task(name="translate_page_subtree")
def do_translate_page_subtree(page_id, locale_ids, components_data, user_id):
    from wagtail.models import Locale
    from wagtail_localize.operations import (
        translate_page_subtree as translate_page_subtree_operation,
    )

    User = get_user_model()

    locales = Locale.objects.filter(pk__in=locale_ids)
    components = pickle.loads(components_data)
    user = User.objects.get(pk=user_id)

    translate_page_subtree_operation(page_id, locales, components, user)

[alexalligator]

Thank you for sharing this code with me, Nigel, and apologies for forgetting to reply. Thanks.!