How does Huey call task?

Question

I have this code here

# it's not every five mins but let's overlook that for now
@periodic_task(crontab(minute='*/1'))
def every_five_mins():
    # ...

But I couldn't find where Huey calls the function. The only other place that I've used Huey is in settings.py but still I only included

HUEY = {
    'huey_class': 'huey.RedisHuey',
    'name': DATABASES['default']['NAME'], 
    'results': True,  
    'store_none': False, 
    'immediate': False, 
    'utc': True,
    'blocking': True,
    'connection': {
        'host': 'localhost',
        'port': 6379,
        'db': 0,
        'connection_pool': None,
        'read_timeout': 1, 
        'url': None, 
    },
    'consumer': {
        'workers': 1,
        'worker_type': 'thread',
        'initial_delay': 0.1,
        'backoff': 1.15, 
        'max_delay': 10.0, 
        'scheduler_interval': 1, 
        'periodic': True,
        'check_worker_health': True,  
        'health_check_interval': 1, 
    },
}

Can anyone please tell me how a task is executed? I want to know this because I want to pass in parameters into every_five_mins(), e.g., every_five_mins(argument1=argument1) but I can't do that without knowing where the function is called (otherwise argument1 is going to raise an undefined error).

Thanks in advance.

Radu · Accepted Answer · 2021-08-12T17:36:04.137

Periodic tasks are called by the consumer (you're running one right?) and I believe the design is such that you aren't meant to pass parameters to these - how would you even pass arguments to the consumer? I'm sure you can come up with a design but to me it doesn't really make sense. From the docs:

Because periodic tasks are called independent of any user interaction, they do not accept any arguments.

Similarly, the return-value for periodic tasks is discarded, rather than being put into the result store. This is because there is not an obvious way for an application to obtain a Result handle to access the result of a given periodic task execution.

Depending on your needs, you may be able to achieve what you want by calling a function that returns some parameters for use within the task:

def get_argument():
    arg = None
    # do stuff
    return arg

# it's not every five mins but let's overlook that for now
@periodic_task(crontab(minute='*/1'))
def every_five_mins():
    argument1 = get_argument()
    # ...

Alternatively, you can also define periodic tasks dynamically. Copying over the example from the docs (note, this doesn't work with process workers -- see link to docs):

def dynamic_ptask(message):
    print('dynamically-created periodic task: "%s"' % message)

@huey.task()
def schedule_message(message, cron_minutes, cron_hours='*'):
    # Create a new function that represents the application
    # of the "dynamic_ptask" with the provided message.
    def wrapper():
        dynamic_ptask(message)

    # The schedule that was specified for this task.
    schedule = crontab(cron_minutes, cron_hours)

    # Need to provide a unique name for the task. There are any number of
    # ways you can do this -- based on the arguments, etc. -- but for our
    # example we'll just use the time at which it was declared.
    task_name = 'dynamic_ptask_%s' % int(time.time())

    huey.periodic_task(schedule, name=task_name)(wrapper)

Assuming the consumer is running, we can now set up as many instances as we like of the “dynamic ptask” function:

>>> from demo import schedule_message
>>> schedule_message('I run every 5 minutes', '*/5')
<Result: task ...>
>>> schedule_message('I run between 0-15 and 30-45', '0-15,30-45')
<Result: task ...>

How does Huey call task?

1 Answers1