2

I'm using dramatiq for my task queue which offers the decorator @dramatiq.actor to decorate a function as a task. I've tried to write my own decorator that wraps the @dramatiq.actor decorator so I can add a default argument to the @dramatiq.actor decorator that applies to all tasks (The argument I'm talking about is priority=100).

For some reason I'm getting the following error:

TypeError: foobar() takes 1 positional argument but 3 were given

If I switch my custom @task decorator with the @dramatiq.actor it works so I guess my custom decorator isn't correct but I can't spot my mistake.

decorators.py

def task(func=None, **kwargs):    
    def decorator(func):
        @wraps(func)
        @dramatiq.actor(priority=100, **kwargs)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)

        return wrapper

    if func is None:
        return decorator

    return decorator(func)

tasks.py

@task
def foobar(entry_pk):
    ...

views.py

foobar.send_with_options(args=(entry.pk,))
Nepo Znat
  • 3,000
  • 5
  • 28
  • 49

2 Answers2

3

It would be much easier to use functools.partial:

from functools import partial

task = partial(dramatiq.actor, priority=100)

@task
def foobar(*args, **kwargs):
    ...

This allows you to preemptively add an argument to a function without actually calling it.

Lord Elrond
  • 13,430
  • 7
  • 40
  • 80
0

Another approach would be to subclass the Dramatiq class and override the actor method. That approach plus some other tips described here - https://blog.narrativ.com/converting-celery-to-dramatiq-a-py3-war-story-23df217b426

Colin Nichols
  • 714
  • 4
  • 12