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

保罗·科尼利厄斯的评论是正确的。您必须像任何其他函数一样调度协程函数：

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

Answer 2

我在使用以下代码时遇到了类似的错误，

RuntimeWarning: coroutine 'process_job' was never awaited

：

async def process_job(job: int):
    await asyncio.sleep(1)
    print(job)

...

scheduler.add_job(lambda: process_job(1), trigger='interval', seconds=5)

就我而言，我使用 asyncio 的

run

函数修复了它

from asyncio import run

async def process_job(job: int):
    await asyncio.sleep(1)
    print(job)


scheduler.add_job(lambda: run(process_job(1)), trigger='interval', seconds=5)

希望这有帮助。