使用fastapi_utilities的repeat_at报错：预期协程，却得到None

错误原因 #

核心问题出在@repeat_at装饰器的行为上：这个装饰器的作用是自动将目标函数注册为FastAPI的后台定时任务，它不会返回原协程函数，装饰后call_api变量指向的是装饰器的返回值（通常是None）。所以当你调用call_api()时，得到的不是协程对象而是None，传入asyncio.create_task()自然就会抛出TypeError: a coroutine was expected, got None。

同时代码中还有两个额外问题：

• 混用repeat_at自动定时和手动create_task触发的逻辑，二者存在冲突；
• call_api内部的while True+break代码完全冗余，无实际作用。

解决方案 #

根据需求分两种场景处理：

场景1：仅需分钟/小时级定时调度 #

直接依赖repeat_at的自动调度能力，FastAPI启动后装饰器会自动按cron规则执行任务，无需手动触发接口。

修改后的代码：

import requests
from fastapi_utilities import repeat_at
from fastapi_offline import FastAPIOffline
import asyncio

app = FastAPIOffline()

# repeat_at装饰器自动注册定时任务，启动后每分钟执行一次
@repeat_at(cron="*/1 * * * *")
async def scheduled_call_api():
    url = "http://sssssss/both/execute_query/4"
    try:
        # 用run_in_executor处理同步的requests请求，避免阻塞事件循环
        loop = asyncio.get_event_loop()
        response = await loop.run_in_executor(None, lambda: requests.get(url))
        response_json = response.json()
        print(response_json)
        return response_json
    except requests.exceptions.RequestException as e:
        print(f"Error fetching data: {e}")

@app.get("/")
async def root():
    return {"message": "Scheduled task is running in background"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=8001)

场景2：需要手动触发+定时调度并存 #

将核心业务逻辑抽离为独立协程函数，避免被repeat_at装饰器覆盖，同时分别实现定时任务和手动触发逻辑。

修改后的代码：

import requests
from fastapi_utilities import repeat_at
from fastapi_offline import FastAPIOffline
import asyncio

app = FastAPIOffline()
manual_task = None

# 独立的核心业务协程，不被装饰器修改
async def call_api():
    url = "http://sssssss/both/execute_query/4"
    try:
        loop = asyncio.get_event_loop()
        response = await loop.run_in_executor(None, lambda: requests.get(url))
        response_json = response.json()
        print(response_json)
        return response_json
    except requests.exceptions.RequestException as e:
        print(f"Error fetching data: {e}")

# 定时任务调用核心函数，每分钟执行一次
@repeat_at(cron="*/1 * * * *")
async def scheduled_task():
    await call_api()

# 手动触发接口
@app.get("/trigger-call-api")
async def trigger_call_api():
    global manual_task
    if manual_task is None or manual_task.done():
        manual_task = asyncio.create_task(call_api())
        return {"message": "call_api triggered manually"}
    else:
        return {"message": "call_api is already running manually"}

# 手动停止接口
@app.get("/stop-call-api")
async def stop_call_api():
    global manual_task
    if manual_task is not None and not manual_task.done():
        manual_task.cancel()
        manual_task = None
        return {"message": "Manual call_api task stopped"}
    else:
        return {"message": "No running manual call_api task"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=8001)

内容的提问来源于stack exchange，提问作者Hosein Sargoli