Я использую 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
.
Итак, мои вопросы,
Как я могу это сделать, это подходящий способ выполнить то, что я пытаюсь сделать?
Как я могу передать *args и **kwargs конкретной функции в словаре tasks
максимально простым способом (чтобы можно было легко добавлять новые задачи) и перенаправить их самой функции задачи через оболочку?
Является ли мой метод периодического повторения функции правильным? Я специально хочу, чтобы он только спал после завершения перед повторным запуском, а не просто снова запускался, даже если последний экземпляр еще не завершился.
@user4815162342 user4815162342 Приношу свои извинения за неясность. Код, который я разместил здесь, представляет собой (слегка) переписанный код, включающий новую функцию-оболочку run_task
, которая должна запускать функции задачи, указанные в словаре tasks
. Я хочу, чтобы некоторые функции периодически повторялись, указав количество секунд с помощью interval
kwarg. Это не работает. Код, который «не работает», является основным блоком кода и 'Task-3': faketask(interval=10)
, который иллюстрирует то, что я пытаюсь сделать.
@user4815162342 user4815162342 Оборачивая его в functools.partial
, похоже, ничего не происходит. Я хочу, чтобы Задача-3 повторялась, но программа просто закрывается.
В вашем исходном вопросе interval
должен был быть необязательным аргументом для faketask
, поэтому я рекомендовал partial
. Здесь вы хотите указать необязательные аргументы для run_task
, поэтому partial
не поможет.
@user4815162342 user4815162342 Я понимаю, что это другая ситуация, и поэтому я разместил это как новый вопрос. Возможно, partial
здесь не решение, но что? Прошу помощи как быть с этим
Вы можете указать дополнительные параметры для run_task
вместе с выполняемой функцией. Например, вместо 'Task-3': faketask(interval=10),
используйте 'Task-3': (faketask, {'interval': 10}),
. При циклическом просмотре элементов словаря распакуйте кортеж как (taskfunc, run_options)
и вызовите run_task(taskname, taskfunc, **run_options)
.
@user4815162342 user4815162342 Я не уверен, что делаю это правильно, но я попытался for taskname, (taskfunc, run_options) in tasks.items():
, и это дало TypeError: cannot unpack non-iterable function object
Вам нужно либо указать их все как кортежи, либо иметь if
, который проверяет, какой вариант вы получили, так.
@user4815162342 user4815162342 Это помогло! Но кажется, что задача не будет повторяться, что немного странно, потому что эта часть должна работать. Он печатает строку, чтобы указать, что она будет повторяться через 10 секунд, но после этого ничего не происходит. В любом случае, огромное спасибо за вашу помощь, я очень рад, что функционал interval
наконец-то работает! Если вы хотите, вы можете опубликовать это как ответ, и я отмечу его как правильный
Исправление: функция повторяется, не отображается только печатное сообщение.
Не потому ли, что напечатанное сообщение не входит в цикл, а находится вне (до) него?
Да, это правильно, мне пришлось отладить с помощью print(), чтобы увидеть это.
Использование 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))