Python - 如何在不阻塞线程的情况下使用FastAPI和uvicorn.run？

Question

我想要在不阻塞线程的情况下，使用uvicorn.run()来运行FastAPI应用程序。我已经尝试过使用进程、子进程和线程，但都没有成功。

我的问题是我想要从另一个进程中启动服务器，在启动服务器后继续执行其他任务。此外，我也遇到了关闭服务器的问题，无法从另一个进程中进行。

是否有人有想法如何使用非阻塞的方式运行uvicorn.run()，以及如何从另一个进程中停止它？

- Leuko

---

此问题的解决方案可以在此答案中找到。 - undefined

Answer 1

由 @HadiAlqattan 提供的方法不起作用，因为 uvicorn.run 希望在主线程中运行。会引发诸如 signal only works in main thread 的错误。

正确的方法是：

import contextlib
import time
import threading
import uvicorn

class Server(uvicorn.Server):
    def install_signal_handlers(self):
        pass

    @contextlib.contextmanager
    def run_in_thread(self):
        thread = threading.Thread(target=self.run)
        thread.start()
        try:
            while not self.started:
                time.sleep(1e-3)
            yield
        finally:
            self.should_exit = True
            thread.join()

config = uvicorn.Config("example:app", host="127.0.0.1", port=5000, log_level="info")
server = Server(config=config)

with server.run_in_thread():
    # Server is started.
    ...
    # Server will be stopped once code put here is completed
    ...

# Server stopped.

使用pytest fixture在本地运行一个实时测试服务器非常方便：

# conftest.py
import pytest

@pytest.fixture(scope="session")
def server():
    server = ...
    with server.run_in_thread():
        yield

制作人：florimondmanca 的 uvicorn#742

- Elijas Dapšauskas

---

您是国王。 - zzfima

我得到了 'Config' 未被定义的错误。它应该是 uvicorn.Config 吗？ - Stealth Rabbi

@StealthRabbi，现在应该已经修复了，谢谢。 - Elijas Dapšauskas

警告：子类化的uvicorn.Server似乎忽略了“workers”配置值，并且只在单个进程中处理请求。 - undefined

实际上，uvicorn.Server是一个单进程设计。它是uvicorn.run实际上利用工作器配置并通过uvicorn.supervisors.Multiprocess启动多个server.run的方式。 - undefined

Answer 2

这是一个备用版本，它可以工作，并且受到了Aponaceuvicorn#1103的启发。Uvicorn维护者希望更多社区参与此问题，因此如果您遇到此问题，请加入讨论。

示例conftest.py文件。

import pytest
from fastapi.testclient import TestClient
from app.main import app
import multiprocessing
from uvicorn import Config, Server


class UvicornServer(multiprocessing.Process):

    def __init__(self, config: Config):
        super().__init__()
        self.server = Server(config=config)
        self.config = config

    def stop(self):
        self.terminate()

    def run(self, *args, **kwargs):
        self.server.run()




@pytest.fixture(scope="session")
def server():
    config = Config("app.main:app", host="127.0.0.1", port=5000, log_level="debug")
    instance = UvicornServer(config=config)
    instance.start()
    yield instance
    instance.stop()

@pytest.fixture(scope="module")
def mock_app(server):
    client = TestClient(app)
    yield client

示例 test_app.py 文件。

def test_root(mock_app):
    response = mock_app.get("")
    assert response.status_code == 200

- polka

---

嗨 @polka，这个方法是有效的，但我有一个问题：如何使用try/except来捕获键盘中断？我的意思是，如果我按下ctrl + c，应用程序会退出，这是可以的（它能识别信号）。然而，我想在退出过程中添加一些日志记录，但我不知道在哪里可以做到这一点。没有异常被抛出，它只是直接退出。你有什么建议吗？ - Tobias

好的，我自己回答了我的问题。你可以将日志消息放在server.run()行下面，因为它不是异步的，所以线程会在那一行被阻塞。如果你按下Ctrl+C，运行会被取消，并执行以下行。 - Tobias

Answer 3

根据Uvicorn文档，没有编程方式来停止服务器。相反，您只能通过按下ctrl + c（官方）来停止服务器。

但是我有一个技巧，可以使用multiprocessing标准库和这三个简单的函数来编程解决此问题：

• 一个运行函数来运行服务器。
• 一个启动函数来启动新进程（启动服务器）。
• 一个停止函数来加入进程（停止服务器）。

from multiprocessing import Process
import uvicorn

# global process variable
proc = None


def run(): 
    """
    This function to run configured uvicorn server.
    """
    uvicorn.run(app=app, host=host, port=port)


def start():
    """
    This function to start a new process (start the server).
    """
    global proc
    # create process instance and set the target to run function.
    # use daemon mode to stop the process whenever the program stopped.
    proc = Process(target=run, args=(), daemon=True)
    proc.start()


def stop(): 
    """
    This function to join (stop) the process (stop the server).
    """
    global proc
    # check if the process is not None
    if proc: 
        # join (stop) the process with a timeout setten to 0.25 seconds.
        # using timeout (the optional arg) is too important in order to
        # enforce the server to stop.
        proc.join(0.25)

有了相同的想法，你可以：

• 使用线程标准库而不是使用多进程标准库。

• 将这些函数重构为一个类。

用法示例：

from time import sleep

if __name__ == "__main__":
    # to start the server call start function.
    start()
    # run some codes ....
    # to stop the server call stop function.
    stop()

您可以了解更多关于：

• Uvicorn 服务器。
• multiprocessing 标准库。
• threading 标准库。
• Concurrency 了解有关Python中多处理和线程的更多信息。

- HadiAlqattan

---

谢谢你的回答，但是你尝试过上面的代码吗？我正在尝试在Win10上使用Python 3.7运行代码，但无论是在线程中启动uvicorn还是在新进程中启动它都会出现错误。在使用线程时，错误看起来像这样：Traceback (most recent call last): File "C:\Python37\lib\site-packages\uvicorn\main.py", line 565, in install_signal_handlers loop.add_signal_handler(sig, self.handle_exit, sig, None) File "C:\Python37\lib\asyncio\events.py", line 540, in add_signal_handler raise NotImplementedError NotImplementedError并且信号只能在主线程中工作。 - Leuko

使用新的进程时出现以下错误：无法pickle _thread.RLock对象。有什么建议可以解决这个问题吗？由于此帖子https://github.com/tiangolo/fastapi/issues/650中提到最好在进程中运行，但对我来说并没有起作用。 - Leuko

好的，我自己找到了解决方案。首先，重要的是使用一个新进程来启动uvicorn。然后，如果你想停止uvicorn，你可以杀死或终止该进程。但是这似乎在Windows上不起作用，至少对我来说只在Linux上起作用。为了避免“无法pickle_thread.RLock对象”的错误，重要的是不要使用带有self的方法。所以例如run_server(self)不能与新进程一起工作，但run_server()可以。 - Leuko

@Leuko发布了一个适当的修复方案，以解决“主线程”错误。 - Elijas Dapšauskas

Answer 4

当我将reload设置为False时，fastapi将启动一个多进程Web服务。如果为True，则Web服务仅有一个进程。

import uvicorn
from fastapi import FastAPI, APIRouter
from multiprocessing import cpu_count
import os

router = APIRouter()
app = FastAPI()


@router.post("/test")
async def detect_img():
    print("pid:{}".format(os.getpid()))
    return os.getpid

if __name__ == '__main__':
    app.include_router(router)
    print("cpu个数：{}".format(cpu_count()))
    workers = 2*cpu_count() + 1
    print("workers:{}".format(workers))
    reload = False
    #reload = True
    uvicorn.run("__main__:app", host="0.0.0.0", port=8082, reload=reload, workers=workers, timeout_keep_alive=5,
                limit_concurrency=100)

- zhoupan0401

Answer 5

我觉得这种异步启动的方式可以解决你的问题。

import asyncio

import uvicorn
from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Hello, World!"}


async def main():
    config = uvicorn.Config(app, port=5000, log_level="info")
    server = uvicorn.Server(config)
    await server.serve()


if __name__ == "__main__":
    asyncio.run(main())

- MOKA xion