Использование планировщика, в котором время хранится в списке, а вход для функции также сохраняется в списке - PullRequest
Новая
       28

Использование планировщика, в котором время хранится в списке, а вход для функции также сохраняется в списке

1 голос
/ 28 апреля 2020

У меня есть два следующих списка: 

marketId_list = ['1.170349308', '1.170349312', '1.170349315']

execution_time_list = ['16:12', '16:13', '16:14']

Я хотел бы перейти к значениям из marketId_list функции в моменты времени, указанные в execution_time_list.

Для вопрос, у меня есть указанная функция, как, например: 

def test(market_identification):
    print(market_identification)

Так что на 16:12 1.170349308 будет напечатано, на 16:13 1.170349312 будет напечатано и на 16:14 1.170349315 будет напечатано.

Для этого я написал следующий код: 

import schedule
import time

for time_of_execution in execution_time_list:

    for market_identification in marketId_list: 

        schedule.every().tuesday.at(time_of_execution).do(test)
        while True:

            schedule.run_pending()
            time.sleep(1)

Хотя это дает мне следующую ошибку: 

TypeError: test() missing 1 required positional argument: 'market_identification'

1 Ответ

2 голосов
/ 28 апреля 2020

Вам необходимо указать отсутствующий аргумент. Одним из способов является использование замыкания , реализованного вложенной функцией.

Здесь при вызове test_wrapper(market_identification) создается новая функция, которая уже имеет необходимый аргумент внутри. Именно эта сгенерированная функция будет затем вызываться каркасом расписания. 


def test(market_identification):
    print(f"test {market_identification}")


def test_wrapper(mi):
    def inner():
        return test(mi)
    return inner


for time_of_execution, market_identification in zip(execution_time_list, marketId_list):
    schedule.every().tuesday.at(time_of_execution).do(test_wrapper(market_identification))
    print(f"scheduled {market_identification} at {time_of_execution}")


while True:
    schedule.run_pending()
    time.sleep(1)

Для простой демонстрации вы можете немедленно вызвать сгенерированную функцию: test_wrapper("hello")() приводит к выводу 

test hello

Поскольку этот вариант использования для вложенной функции очень распространен, при использовании метода functools.partial стандартная библиотека предоставляет вам небольшой помощник для этого: 

from functools import partial

for time_of_execution, market_identification in zip(execution_time_list, marketId_list):
    schedule.every().tuesday.at(time_of_execution).do(partial(test, market_identification))
    print(f"scheduled {market_identification} at {time_of_execution}")

Обратите внимание, что while l oop не должен находиться во внутреннем l oop. Это должно быть выполнено после того, как все планировщики были установлены. Я предполагаю, что отступ был испорчен в исходном коде. Это исправлено в моем ответе выше.

Обновление: Сначала я пропустил ту часть, где вы объясняете, что каждый раз должен использоваться только один идентификатор рынка. В этом случае вам нужны не два вложенных цикла, а один l oop, который zip s * execution_time_list и marketId_list. Я обновил свой ответ соответственно.

...