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

Я ищу возможность использовать uvicorn.run() с приложением FstAPI, но без uvicorn.run() блокирует поток. Я уже пробовал использовать процессы, подпроцессы и потоки, но ничего не получалось. Моя проблема в том, что я хочу запустить сервер из другого процесса, который должен продолжить выполнение других задач после запуска сервера. Кроме того, у меня возникают проблемы с закрытием сервера таким образом из другого процесса.

Кто-нибудь знает, как использовать uvicorn.run() без блокировки и как остановить его от другого процесса?

Привет LeukoClassic

Источник

5 ответов

Решение

Согласно документации Uvicorn, программно остановить сервер невозможно. вместо этого вы можете остановить сервер, только нажав ctrl + c (официально).

Но у меня есть трюк, чтобы решить эту проблему программно, используя стандартную многопроцессорную библиотеку с этими тремя простыми функциями:

  • Функция запуска для запуска сервера.
  • Функция запуска для запуска нового процесса (запуск сервера).
  • Функция остановки для присоединения к процессу (остановка сервера).
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()



Вы можете узнать больше о:

Источник

Подход, предложенный @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 от https://github.com/florimondmanca

Источник

Это альтернативная версия, которая работает и была вдохновлена ​​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
Источник

Когда я устанавливаю для перезагрузки значение False, fastapi запускает веб-службу с несколькими процессами. Если это правда, для веб-службы будет только один процесс.

      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)
Источник

Я думаю, что такой асинхронный стартап может решить вашу проблему.

      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())
Источник
Другие вопросы по тегам