Huey如何调用任务？

Question

我这里有这个代码

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

但我找不到 Huey 调用该函数的位置。我用过 Huey 的唯一其他地方是 settings.py，但我仍然只包括

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, 
    },
}

谁能告诉我任务是如何执行的？我想知道这一点，因为我想将参数传递给 every_five_mins()，例如 every_five_mins(argument1=argument1)，但我不能在不知道调用函数的位置的情况下这样做（否则 argument1 将引发未定义的错误).

提前致谢。

Answer 1

周期性任务由消费者调用（您是运行之一，对吗？）并且我相信设计是这样的，您不打算将参数传递给这些 - 您甚至如何传递参数给消费者？我相信你可以想出一个设计，但对我来说它并没有真正的意义。来自 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.

根据您的需要，您可以通过调用一个函数来实现您想要的效果，该函数returns 一些参数在任务中使用：

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()
    # ...

或者，您也可以 define periodic tasks dynamically。从文档中复制示例（注意，这不适用于流程工作人员——请参阅 link 文档）：

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 ...>

Huey如何调用任务？

How does Huey call task?

django

python-huey