我想要在不阻塞线程的情况下,使用uvicorn.run()来运行FastAPI应用程序。我已经尝试过使用进程、子进程和线程,但都没有成功。
我的问题是我想要从另一个进程中启动服务器,在启动服务器后继续执行其他任务。此外,我也遇到了关闭服务器的问题,无法从另一个进程中进行。
是否有人有想法如何使用非阻塞的方式运行uvicorn.run(),以及如何从另一个进程中停止它?
我想要在不阻塞线程的情况下,使用uvicorn.run()来运行FastAPI应用程序。我已经尝试过使用进程、子进程和线程,但都没有成功。
我的问题是我想要从另一个进程中启动服务器,在启动服务器后继续执行其他任务。此外,我也遇到了关闭服务器的问题,无法从另一个进程中进行。
是否有人有想法如何使用非阻塞的方式运行uvicorn.run(),以及如何从另一个进程中停止它?
由 @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
uvicorn.Config
吗? - Stealth Rabbi示例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
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()
您可以了解更多关于:
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())