Python

Question

我正在寻找将 uvicorn.run() 与 FastAPI 应用程序一起使用但没有 uvicorn.run() 阻塞线程的可能性。我已经尝试使用进程、子进程和线程，但没有任何效果。我的问题是我想从另一个应该在启动服务器后继续执行其他任务的进程启动服务器。另外，我在从另一个进程关闭服务器时遇到了问题。

有谁知道如何使用 uvicorn.run() 非阻塞以及如何从另一个进程中停止它？

Answer 1

根据 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)

有了同样的想法，你可以 :

使用threading标准库而不是使用multiprocessing标准库。
将这些函数重构为class.

使用示例：

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 中了解有关多处理和线程的更多信息。

Answer 2

@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 = 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

学分：uvicorn#742 by florimondmanca

Answer 3

这是另一个可用的版本，其灵感来自 Aponace uvicorn#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

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

Python - How to use FastAPI and uvicorn.run without blocking the thread?

multiprocessing

fastapi

uvicorn