FastAPI Python:如何在后台运行线程?

xv8emn3q  于 2023-01-29  发布在  Python
关注(0)|答案(2)|浏览(673)

我正在用FastAPI在python中做一个服务器,我希望有一个与我的API无关的函数,每5分钟在后台运行一次(比如从API中检查内容,并根据响应打印内容)
我试过创建一个线程来运行start_worker函数,但是它什么也不打印。
有人知道怎么做吗?

def start_worker():
    print('[main]: starting worker...')
    my_worker = worker.Worker()
    my_worker.working_loop() # this function prints "hello" every 5 seconds

if __name__ == '__main__':
    print('[main]: starting...')
    uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)
    _worker_thread = Thread(target=start_worker, daemon=False)
    _worker_thread.start()
cidc1ykv

cidc1ykv1#

您应该在调用uvicorn.run之前启动Thread,因为uvicorn.run正在阻塞该线程。
PS:在你的问题中你说你希望后台任务每5分钟运行一次,但是在你的代码中你说每5秒运行一次。下面的例子假设你希望是后者。如果你希望它每5分钟运行一次,那么把时间调整为60 * 5**。

    • 备选案文1**
import time
import threading
from fastapi import FastAPI
import uvicorn

app = FastAPI()
class BackgroundTasks(threading.Thread):
    def run(self,*args,**kwargs):
        while True:
            print('Hello')
            time.sleep(5)
  
if __name__ == '__main__':
    t = BackgroundTasks()
    t.start()
    uvicorn.run(app, host="0.0.0.0", port=8000)

您还可以使用FastAPI的启动事件启动线程,只要它可以在应用程序启动之前运行即可。

@app.on_event("startup")
async def startup_event():
    t = BackgroundTasks()
    t.start()
    • 备选案文2**

您可以改为使用重复的Event scheduler执行后台任务,如下所示:

import sched, time
from threading import Thread
from fastapi import FastAPI
import uvicorn

app = FastAPI()
s = sched.scheduler(time.time, time.sleep)

def print_event(sc): 
    print("Hello")
    sc.enter(5, 1, print_event, (sc,))

def start_scheduler():
    s.enter(5, 1, print_event, (s,))
    s.run()

@app.on_event("startup")
async def startup_event():
    thread = Thread(target = start_scheduler)
    thread.start()

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

68bkxrlz2#

@Chris的解决方案工作正常!
但是,如果您想改进它并删除冗余的start_scheduler方法,只需将sc参数直接传递给print_event方法,kwargs参数如下所示:

import sched
import time
from threading import Thread

import uvicorn
from fastapi import FastAPI

app = FastAPI()
s = sched.scheduler(time.time, time.sleep)

def print_event(sc=None): 
    print("Hello")
    sc.enter(5, 1, print_event, (sc,))
    sc.run()

@app.on_event("startup")
async def startup_event():
    thread = Thread(target=print_event, kwargs=dict(sc=s))
    thread.start()

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

相关问题