如何让FastAPI在主循环运行时访问ConnectionCheckEngine的数据？

你当前的写法行不通——因为cce.run()里的while True死循环会直接阻塞主线程，导致FastAPI的Web服务根本没机会启动。必须把数据采集的循环放到独立的线程/异步任务里，让它和FastAPI服务并行运行。

方法1：使用多线程 #

把ConnectionCheckEngine的run方法放到子线程中执行，主线程可以正常启动FastAPI服务：

import time
import threading
from fastapi import FastAPI

class ConnectionCheckEngine():
    def __init__(self):
       self.data = list()

    def get_connection_info(self):
        # 替换为你实际获取SSH连接信息的逻辑
        return {"timestamp": time.time(), "connections": []}

    def run(self):
       while True:
           self.data.append(self.get_connection_info())
           time.sleep(30)

app = FastAPI()
cce = ConnectionCheckEngine()

@app.get("/")
async def root():
    return {'data': cce.data}

if __name__ == '__main__':
    # 启动数据采集守护线程，主线程退出时自动终止
    check_thread = threading.Thread(target=cce.run, daemon=True)
    check_thread.start()
    # 启动FastAPI服务，需提前安装uvicorn：pip install uvicorn
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

• 列表的append操作在Python中是线程安全的，当前场景无需额外加锁；如果后续采集逻辑涉及复杂共享资源操作，需使用threading.Lock保证线程安全。

方法2：使用异步任务（适配FastAPI异步生态） #

如果get_connection_info可以改造成异步方法，推荐用异步任务运行采集循环：

import asyncio
from fastapi import FastAPI

class ConnectionCheckEngine():
    def __init__(self):
       self.data = list()

    async def get_connection_info(self):
        # 替换为异步的SSH连接信息采集逻辑
        return {"timestamp": asyncio.get_event_loop().time(), "connections": []}

    async def run(self):
       while True:
           self.data.append(await self.get_connection_info())
           await asyncio.sleep(30)

app = FastAPI()
cce = ConnectionCheckEngine()

@app.on_event("startup")
async def startup_event():
    # FastAPI启动时自动启动后台异步任务
    asyncio.create_task(cce.run())

@app.get("/")
async def root():
    return {'data': cce.data}

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

• 这种方式更适合FastAPI的异步架构，尤其当采集操作涉及IO等待（如SSH连接）时，能更高效利用系统资源。

额外注意事项 #

• 若self.data持续增长，建议定期清理旧数据，避免内存溢出；
• 若后续涉及对self.data的复杂修改操作，需确保线程/异步安全。

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