Как использовать FastAPI и uvicorn.run без блокировки потока?

Согласно документации 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 стандартная библиотека.

• преобразовать эти функции в класс.

Пример использования:

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 стандартной библиотеки.
• Параллелизм, чтобы узнать больше о многопоточности и многопоточности в Python.

Подход, предложенный @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:

# conftest.py
import pytest

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

Кредиты: uvicorn # 742 от florimondmanca