APScheduler 调用异步函数

Question

我一直在努力安排一个异步函数。我的理解是 AsyncIOScheduler 会让我这样做，但在这个特定的代码块中，我没有得到任何快乐。

简单地运行一个基本的例子没问题:

from datetime import datetime 
import asyncio
from apscheduler.schedulers.asyncio import AsyncIOScheduler

async def tick():
    print(f"Tick! The async time is {datetime.now()}")

if __name__ == '__main__':
    scheduler = AsyncIOScheduler()
    scheduler.add_job(tick, 'interval', seconds=3)
    scheduler.start()

    try:
        asyncio.get_event_loop().run_forever()
    except (KeyboardInterrupt, SystemExit):
        pass

生产：

Tick! The async time is 2021-07-01 18:47:43.503460
Tick! The async time is 2021-07-01 18:47:46.500421
Tick! The async time is 2021-07-01 18:47:49.500208
^C%

然而，我的代码块因有关可调用项的各种事情而出错，并继续告诉我从未等待 seed。它等待 main() 在设置调度程序之前很好地调用，但也许这是一个转移注意力的问题？

我不是python专家，所以请稍微温和一点:)

提前感谢您提供任何线索。

import asyncio
from apscheduler.schedulers.asyncio import AsyncIOScheduler
import c_seed
import c_logging


client = 'async'
state = 'live'
account = 'spot'

async def main():
    c_logging.logging.info("")
    c,l = await c_seed.init(client,state,account)
    await c_seed.seed(c,l)
    scheduler = AsyncIOScheduler()
    scheduler.add_job(c_seed.seed(c,l), 'interval', seconds=60)
    scheduler.start()

if __name__ == "__main__":
    c_logging.logging.info("")
    asyncio.run(main())

2021-07-01 18:43:59,943 [INFO] <module>: c_client
2021-07-01 18:43:59,944 [INFO] <module>: 
2021-07-01 18:43:59,945 [DEBUG] __init__: Using selector: KqueueSelector
2021-07-01 18:43:59,945 [INFO] main: 
2021-07-01 18:43:59,945 [INFO] init: 
2021-07-01 18:43:59,945 [INFO] create: c_client
2021-07-01 18:43:59,945 [INFO] create: c_client -> async
2021-07-01 18:43:59,945 [INFO] create: c_client -> live
2021-07-01 18:44:00,591 [INFO] pairs_list: 
2021-07-01 18:44:00,753 [DEBUG] seed: async seed start
2021-07-01 18:44:07,040 [DEBUG] seed: async seed complete
Traceback (most recent call last):
  File "/Users/Documents/Development/main.py", line 58, in <module>
    asyncio.run(main()) 
  File "/Applications/Xcode.app/Contents/Developer/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "/Applications/Xcode.app/Contents/Developer/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete
    return future.result()
  File "/Users/Documents/Development/main.py", line 36, in main
    scheduler.add_job(c_seed.seed(c,l), 'interval', seconds=60)
  File "/Users/Documents/Development/.venv/lib/python3.8/site-packages/apscheduler/schedulers/base.py", line 439, in add_job
    job = Job(self, **job_kwargs)
  File "/Users/Documents/Development/.venv/lib/python3.8/site-packages/apscheduler/job.py", line 49, in __init__
    self._modify(id=id or uuid4().hex, **kwargs)
  File "/Users/Documents/Development/.venv/lib/python3.8/site-packages/apscheduler/job.py", line 170, in _modify
    raise TypeError('func must be a callable or a textual reference to one')
TypeError: func must be a callable or a textual reference to one
sys:1: RuntimeWarning: coroutine 'seed' was never awaited

Answer 1

Paul Cornelius 的评论是正确的。您必须像任何其他函数一样安排协程函数：

scheduler.add_job(c_seed.seed, 'interval', seconds=60, args=(c,l))

APScheduler 调用异步函数

APScheduler to call an async function

python

apscheduler