celery_django.py

def pk_or_instance(fn):
    try:
        first_param_name = next(iter(inspect.signature(fn).parameters))
    except StopIteration:
        raise ValueError(f"Function {fn} needs to take at least one parameter")

    type_hints = get_type_hints(fn, fn.__globals__)
    model_type = type_hints.get(first_param_name)
    if not model_type:
        raise ValueError(
            f"Parameter {first_param_name} did not have a type annotation"
        )

    try:
        is_model = issubclass(model_type, Model)
    except TypeError:
        is_model = False

    if not is_model:
        raise ValueError(
            f"Parameter {first_param_name}'s type annotation was not for a Django Model"
        )

    @functools.wraps(fn)
    def inner(pk, *args, **kwargs):
        if not isinstance(pk, model_type):
            pk = model_type.objects.get(pk=pk)
        return fn(pk, *args, **kwargs)

    return inner
  
  
# todo create a custom Task subclass that auto-serializes model arguments to .apply_async into their PKs