Передача *args и **kwargs в асинхронную периодическую функцию-оболочку из словаря

Я использую asyncio для сбора задач из словаря и их выполнения, но мне трудно заставить его работать, как задумано. (Это своего рода дополнительный вопрос к моему вопросу здесь, но я немного переписал код, потому что он не работал должным образом, и решил, что вместо этого лучше использовать функцию-оболочку.)

Поэтому я использую оболочку для вызова указанной функции задачи. Я хочу, чтобы оболочка перенаправляла любые *args или **kwargs в функцию задачи, а также периодически повторяла задачу, если установлен interval kwarg.

Как мне передать эту информацию оболочке и функции задачи, сохраняя при этом легкость обслуживания с возможностью легко добавлять новые задачи в словарь tasks?

Пожалуйста, взгляните на мой код для иллюстрации.

import asyncio
import random

async def run_task(taskname, taskfunc, interval=None, *args, **kwargs):
    # Wrapper which will run the specified function, and repeat it if 'interval' is set.
    # Should also be able to pass any potential *args and **kwargs to the function.
    fakedelay = random.randint(1,6)
    print(f'{taskname} started (completing in {fakedelay} seconds)')
    await taskfunc(fakedelay, *args, **kwargs)
    print(f'{taskname} completed after {fakedelay} seconds')
    if interval is not None:
        print(f'Repeating {taskname} in {interval} seconds...')
        while True:
            await taskfunc(fakedelay, *args, **kwargs)
            await asyncio.sleep(interval)

async def faketask(fakedelay, *args, **kwargs):
    # Function to simulate a coroutine task
    await asyncio.sleep(fakedelay)

async def main():
    tasks = {
        # Dictionary of tasks to perform
        'Task-1': faketask,
        'Task-2': faketask,
        'Task-3': faketask,
    }

    tasklist = []
    for taskname, taskfunc in tasks.items():
        tasklist.append(run_task(taskname, taskfunc))
        print(f'Added {taskname} to job queue.')
    await asyncio.gather(*tasklist)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.close()

Кажется, пока это работает хорошо. Но допустим, что я хочу, чтобы Задача-3 повторялась каждые 10 секунд после каждого завершения. Я хотел бы просто указать это в словаре tasks, чтобы было максимально просто добавлять новые задачи в будущем. Например. так:

tasks = {
    # Dictionary of tasks to perform
    'Task-1': faketask,
    'Task-2': faketask,
    'Task-3': faketask(interval=10),
}

Но запуск этого дает TypeError: faketask() missing 1 required positional argument: 'fakedelay' Я полагаю, это имеет смысл, потому что interval kwarg предназначен для оболочки, а не самой функции задачи (faketask). И оболочка, похоже, не может добавить никаких *args или **kwargs (fakedelay в этой ситуации).

В моем предыдущем вопросе мне предложили использовать functools.partial.

tasks = {
    'Task-1': faketask,
    'Task-2': faketask,
    'Task-3': functools.partial(faketask, interval=10),
}

Это несколько решило проблему из моего предыдущего вопроса, но после переписывания кода и добавления функции-оболочки теперь он, похоже, ничего не делает, и, по общему признанию, мне трудно понять, как должен использоваться functools.partial.

Итак, мои вопросы,

  1. Как я могу это сделать, это подходящий способ выполнить то, что я пытаюсь сделать?

  2. Как я могу передать *args и **kwargs конкретной функции в словаре tasks максимально простым способом (чтобы можно было легко добавлять новые задачи) и перенаправить их самой функции задачи через оболочку?

  3. Является ли мой метод периодического повторения функции правильным? Я специально хочу, чтобы он только спал после завершения перед повторным запуском, а не просто снова запускался, даже если последний экземпляр еще не завершился.

после переписывания кода и добавления функции-оболочки, теперь он, похоже, ничего не делает - покажите переписанный код. Вопрос длинный, но на самом деле он не включает код, который дает сбой!
user4815162342 08.04.2019 08:23

@user4815162342 user4815162342 Приношу свои извинения за неясность. Код, который я разместил здесь, представляет собой (слегка) переписанный код, включающий новую функцию-оболочку run_task, которая должна запускать функции задачи, указанные в словаре tasks. Я хочу, чтобы некоторые функции периодически повторялись, указав количество секунд с помощью interval kwarg. Это не работает. Код, который «не работает», является основным блоком кода и 'Task-3': faketask(interval=10), который иллюстрирует то, что я пытаюсь сделать.

noob 08.04.2019 08:42

@user4815162342 user4815162342 Оборачивая его в functools.partial, похоже, ничего не происходит. Я хочу, чтобы Задача-3 повторялась, но программа просто закрывается.

noob 08.04.2019 08:48

В вашем исходном вопросе interval должен был быть необязательным аргументом для faketask, поэтому я рекомендовал partial. Здесь вы хотите указать необязательные аргументы для run_task, поэтому partial не поможет.

user4815162342 08.04.2019 08:51

@user4815162342 user4815162342 Я понимаю, что это другая ситуация, и поэтому я разместил это как новый вопрос. Возможно, partial здесь не решение, но что? Прошу помощи как быть с этим

noob 08.04.2019 08:54

Вы можете указать дополнительные параметры для run_task вместе с выполняемой функцией. Например, вместо 'Task-3': faketask(interval=10), используйте 'Task-3': (faketask, {'interval': 10}),. При циклическом просмотре элементов словаря распакуйте кортеж как (taskfunc, run_options) и вызовите run_task(taskname, taskfunc, **run_options).

user4815162342 08.04.2019 09:02

@user4815162342 user4815162342 Я не уверен, что делаю это правильно, но я попытался for taskname, (taskfunc, run_options) in tasks.items():, и это дало TypeError: cannot unpack non-iterable function object

noob 08.04.2019 09:23

Вам нужно либо указать их все как кортежи, либо иметь if, который проверяет, какой вариант вы получили, так.

user4815162342 08.04.2019 11:10

@user4815162342 user4815162342 Это помогло! Но кажется, что задача не будет повторяться, что немного странно, потому что эта часть должна работать. Он печатает строку, чтобы указать, что она будет повторяться через 10 секунд, но после этого ничего не происходит. В любом случае, огромное спасибо за вашу помощь, я очень рад, что функционал interval наконец-то работает! Если вы хотите, вы можете опубликовать это как ответ, и я отмечу его как правильный

noob 08.04.2019 11:43

Исправление: функция повторяется, не отображается только печатное сообщение.

noob 08.04.2019 12:35

Не потому ли, что напечатанное сообщение не входит в цикл, а находится вне (до) него?

user4815162342 08.04.2019 12:40

Да, это правильно, мне пришлось отладить с помощью print(), чтобы увидеть это.

noob 08.04.2019 12:43
Почему в Python есть оператор "pass"?
Почему в Python есть оператор "pass"?
Оператор pass в Python - это простая концепция, которую могут быстро освоить даже новички без опыта программирования.
Некоторые методы, о которых вы не знали, что они существуют в Python
Некоторые методы, о которых вы не знали, что они существуют в Python
Python - самый известный и самый простой в изучении язык в наши дни. Имея широкий спектр применения в области машинного обучения, Data Science,...
Основы Python Часть I
Основы Python Часть I
Вы когда-нибудь задумывались, почему в программах на Python вы видите приведенный ниже код?
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
Алиса и Боб имеют неориентированный граф из n узлов и трех типов ребер:
Оптимизация кода с помощью тернарного оператора Python
Оптимизация кода с помощью тернарного оператора Python
И последнее, что мы хотели бы показать вам, прежде чем двигаться дальше, это
Советы по эффективной веб-разработке с помощью Python
Советы по эффективной веб-разработке с помощью Python
Как веб-разработчик, Python может стать мощным инструментом для создания эффективных и масштабируемых веб-приложений.
0
12
427
1
Перейти к ответу Данный вопрос помечен как решенный

Ответы 1

Ответ принят как подходящий

Использование functools.partial имеет смысл только в том случае, если вы на самом деле оборачиваете faketask, чтобы включить необязательный аргумент ключевого слова. Если вам нужно применить аргумент ключевого слова к другой функции (run_task), вам нужно сделать это независимо. Например, вы можете указать дополнительные опции для run_task в tasks dict:

tasks = {
    'Task-1': faketask,
    'Task-2': faketask,
    'Task-3': (faketask, {'interval': 10)),
}

Затем код, который вызывает run_task, должен будет распознать кортежи:

for taskname, taskfunc_maybe_with_options in tasks.items():
    if isinstance(taskfunc_maybe_with_options, tuple):
        taskfunc, options = taskfunc_maybe_with_options
    else:
        taskfunc = taskfunc_maybe_with_options
        options = {}
    tasklist.append(run_task(taskname, taskfunc, **options))

Другие вопросы по теме