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

2024-12-11 08:47:00
admin
原创
229
摘要:问题描述:我正在使用 FastAPI 在 python 中创建一个服务器,我想要一个与我的 API 无关的函数,每 5 分钟在后台运行一次(比如从 API 检查内容并根据响应打印内容)我试图创建一个运行该函数的线程start_worker,但它没有打印任何内容。有人知道怎么做吗?def start_worke...

问题描述:

我正在使用 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()

解决方案 1:

选项 1

您应该Thread在调用之前启动您的uvicorn.run,因为它uvicorn.run会阻塞线程。

from fastapi import FastAPI
import threading
import uvicorn
import time

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)

您也可以Thread使用 FastAPI 的启动事件来启动,只要它可以在应用程序启动之前运行即可(更新:请参阅下面的选项 3 了解如何使用lifespan事件,因为startup事件现在已被弃用)。

@app.on_event("startup")
async def startup_event():
    t = BackgroundTasks()
    t.start()

选项 2

除了while True:循环之外,您还可以使用重复事件调度程序来执行后台任务,如下所示:

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

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)

选项 3

如果您的任务是一个async def函数(有关 FastAPI 中的vs端点/后台任务的更多详细信息,请参阅此答案),那么您可以使用该函数将任务添加到当前事件循环。该函数接受一个协程对象(即一个函数)并返回一个对象(如果需要,可以将其用于任务,或者它等)。该调用在当前线程的事件循环内创建任务,并与事件循环中的所有其他任务同时在“后台”执行它,并在它们之间切换。def`async defasyncio.create_task()create_task()async defTaskawaitcancel`await

在调用 之前,需要创建一个事件循环,并且该循环在以编程方式(例如,使用)或在终端中(例如,使用)create_task()启动服务器时已经创建。除了使用,还可以使用获取当前事件,然后调用。uvicorn`uvicorn.run(app)uvicorn app:appasyncio.create_task()asyncio.get_running_loop()loop`loop.create_task()

下面的示例使用最近记录的添加事件的方法lifespan(使用上下文管理器),即在应用程序启动之前以及应用程序关闭时应执行的代码(请参阅文档,以及此答案和此答案以获取更多详细信息和示例)。 也可以继续使用startupshutdown事件,如前面的选项所示;但是,这些事件处理程序可能会从未来的 FastAPI/Starlette 版本中删除。

from fastapi import FastAPI
from contextlib import asynccontextmanager
import asyncio


async def print_task(s): 
    while True:
        print('Hello')
        await asyncio.sleep(s)

        
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Run at startup
    asyncio.create_task(print_task(5))
    yield
    # Run on shutdown (if required)
    print('Shutting down...')


app = FastAPI(lifespan=lifespan)

其他解决方案

替代解决方案可能包括使用ApScheduler,更具体地说AsyncIOScheduler,如本答案所示。

相关推荐
  政府信创国产化的10大政策解读一、信创国产化的背景与意义信创国产化,即信息技术应用创新国产化,是当前中国信息技术领域的一个重要发展方向。其核心在于通过自主研发和创新,实现信息技术应用的自主可控,减少对外部技术的依赖,并规避潜在的技术制裁和风险。随着全球信息技术竞争的加剧,以及某些国家对中国在科技领域的打压,信创国产化显...
工程项目管理   1590  
  为什么项目管理通常仍然耗时且低效?您是否还在反复更新电子表格、淹没在便利贴中并参加每周更新会议?这确实是耗费时间和精力。借助软件工具的帮助,您可以一目了然地全面了解您的项目。如今,国内外有足够多优秀的项目管理软件可以帮助您掌控每个项目。什么是项目管理软件?项目管理软件是广泛行业用于项目规划、资源分配和调度的软件。它使项...
项目管理软件   1361  
  信创产品在政府采购中的占比分析随着信息技术的飞速发展以及国家对信息安全重视程度的不断提高,信创产业应运而生并迅速崛起。信创,即信息技术应用创新,旨在实现信息技术领域的自主可控,减少对国外技术的依赖,保障国家信息安全。政府采购作为推动信创产业发展的重要力量,其对信创产品的采购占比情况备受关注。这不仅关系到信创产业的发展前...
信创和国产化的区别   18  
  信创,即信息技术应用创新产业,旨在实现信息技术领域的自主可控,摆脱对国外技术的依赖。近年来,国货国用信创发展势头迅猛,在诸多领域取得了显著成果。这一发展趋势对科技创新产生了深远的推动作用,不仅提升了我国在信息技术领域的自主创新能力,还为经济社会的数字化转型提供了坚实支撑。信创推动核心技术突破信创产业的发展促使企业和科研...
信创工作   18  
  信创技术,即信息技术应用创新产业,旨在实现信息技术领域的自主可控与安全可靠。近年来,信创技术发展迅猛,对中小企业产生了深远的影响,带来了诸多不可忽视的价值。在数字化转型的浪潮中,中小企业面临着激烈的市场竞争和复杂多变的环境,信创技术的出现为它们提供了新的发展机遇和支撑。信创技术对中小企业的影响技术架构变革信创技术促使中...
信创国产化   19  
热门文章
项目管理软件有哪些?
云禅道AD
禅道项目管理软件

云端的项目管理软件

尊享禅道项目软件收费版功能

无需维护,随时随地协同办公

内置subversion和git源码管理

每天备份,随时转为私有部署

免费试用