Generic Celery task that calls Django model methods asynchronously

tasks.py

import importlib
from functools import wraps
from myproject.celery import app


@app.task
def call_async_task(obj_module_name, obj_class_name, obj_pk, obj_method, obj_args=None, obj_kwargs=None):
    model_class = getattr(importlib.import_module(obj_module_name), obj_class_name)
    obj = model_class.objects.get(pk=obj_pk)
    method = getattr(obj, obj_method)
    obj_args = obj_args or []
    obj_kwargs = obj_kwargs or {}
    return method(*obj_args, **obj_kwargs)


def call_async(obj, obj_method, *obj_args, **obj_kwargs):
    call_async_task.delay(obj.__module__, obj.__class__.__name__, obj.pk, obj_method, obj_args, obj_kwargs)


def async_capable(f):
    @wraps(f)
    def func(self, *args, **kwargs):
        if kwargs.pop('async', False):
            return call_async(self, func.__name__, *args, **kwargs)
        return f(self, *args, **kwargs)
    return func