FastAPI中AsyncIO任务无法记录日志或打印内容的原因排查

我遇到的问题和其他同类问题不同——我专门询问为何通过AsyncIO loop运行任务时无法生成日志，其他问题只讲如何运行AsyncIO任务，没涉及日志失效的情况。
我用新线程运行长时间任务，Celery/BackgroundTasks不适合我的场景，所以采用了线程运行方案。尝试用Logger输出到文件，但被调用的长时间函数既无法输出日志，print语句也失效，现在连函数有没有运行都没法判断。

代码如下：

import logging
import asyncio
from fastapi import FastAPI
from time import sleep

app = FastAPI()

class WorkerLogger:
    def __init__(self) -> None:
        self.logger = logging.getLogger('worker')
        handler = logging.FileHandler('worker.logs')
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)

    def log(self, message: str) -> None:
        self.logger.info(message)

worker_logger = WorkerLogger()

async def long_running_task():
    worker_logger.log("❌ this does not get logged")
    sleep(5)

# Root route
@app.get("/")
async def root():
    loop = asyncio.get_running_loop()
    worker_logger.log("✅ this will get logged")
    newdata = await loop.run_in_executor(None, lambda: long_running_task())
    return {"result": "true"}

问题根源 #

核心问题是你用loop.run_in_executor调用异步函数时，只创建了协程对象但没有执行它。long_running_task是用async定义的异步函数，你通过lambda: long_running_task()返回的只是一个未被调度执行的协程对象，所以函数内部的日志、sleep逻辑完全没触发，自然看不到任何输出。

解决办法 #

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

场景1：不需要long_running_task是异步函数 #

既然函数里用的是同步的sleep，完全可以把它改成普通同步函数，这样run_in_executor能直接执行：

# 修改任务函数为同步函数
def long_running_task():
    worker_logger.log("✅ 现在能正常输出日志了")
    sleep(5)

# 路由里直接传函数名，无需lambda
@app.get("/")
async def root():
    loop = asyncio.get_running_loop()
    worker_logger.log("✅ this will get logged")
    newdata = await loop.run_in_executor(None, long_running_task)
    return {"result": "true"}

场景2：必须保留long_running_task为异步函数 #

如果任务里确实有异步逻辑需要保留async定义，那么需要在子线程中手动创建并运行asyncio事件循环来执行协程：

# 封装一个同步函数来运行异步任务
def run_async_task():
    import asyncio
    asyncio.run(long_running_task())

# 路由里调用这个封装函数
@app.get("/")
async def root():
    loop = asyncio.get_running_loop()
    worker_logger.log("✅ this will get logged")
    newdata = await loop.run_in_executor(None, run_async_task)
    return {"result": "true"}

额外调试建议 #

可以给WorkerLogger添加控制台输出Handler，方便实时查看日志，避免只写文件难以调试：

class WorkerLogger:
    def __init__(self) -> None:
        self.logger = logging.getLogger('worker')
        # 文件输出Handler
        file_handler = logging.FileHandler('worker.logs')
        file_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        file_handler.setFormatter(file_formatter)
        # 控制台输出Handler
        console_handler = logging.StreamHandler()
        console_formatter = logging.Formatter('%(levelname)s: %(message)s')
        console_handler.setFormatter(console_formatter)
        
        self.logger.addHandler(file_handler)
        self.logger.addHandler(console_handler)
        self.logger.setLevel(logging.INFO)

内容的提问来源于stack exchange，提问作者Tejas Sharma