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

Question

I'm looking for a possibility to use uvicorn.run() with a FastAPI app but without uvicorn.run() is blocking the thread. I already tried to use processes, subprocessesand threads but nothing worked. My problem is that I want to start the Server from another process that should go on with other tasks after starting the server. Additinally I have problems closing the server like this from another process.

Has anyone an idea how to use uvicorn.run() non blocking and how to stop it from another process?

Answer 1

According to Uvicorn documentation there is no programmatically way to stop the server. instead, you can stop the server only by pressing ctrl + c (officially).

But I have a trick to solve this problem programmatically using multiprocessing standard lib with these three simple functions :

• A run function to run the server.
• A start function to start a new process (start the server).
• A stop function to join the process (stop the server).

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)

With the same idea you can :

• use threading standard lib instead of using multiprocessing standard lib.

• refactor these functions into a class.

Example of usage :

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()

You can read more about :

• Uvicorn server.
• multiprocessing standard lib.
• threading standard lib.
• Concurrency to know more about multi processing and threading in python.

Answer 2

Approach given by @HadiAlqattan will not work because uvicorn.run expects to be run in the main thread. Errors such as signal only works in main thread will be raised.

Correct approach is:

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.

Very handy to run a live test server locally using a pytest fixture:

# conftest.py
import pytest

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

Credits: uvicorn#742 by florimondmanca

Answer 3

This is an alternate version which works and was inspired by Aponace uvicorn#1103. The uvicorn maintainers want more community engagement with this issue, so if you are experiencing it, please join the conversation.

Example conftest.py file.

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

Example test_app.py file.

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