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?

Greetings LeukoClassic

Solution 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.

Solution 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

Comments

[Leuko]

Thanks for the answer but did you try your code above? I'm trying to run the code on Win10 with python 3.7 and I get errors either starting uvicorn in a Thread or starting it in a new process. The error using a thread looks like this: Traceback (most recent call last): File "C:\Python37\lib\site-packages\uvicorn\main.py", line 565, in install_signal_handlers loop.add_signal_handler(sig, self.handle_exit, sig, None) File "C:\Python37\lib\asyncio\events.py", line 540, in add_signal_handler raise NotImplementedError NotImplementedError and signal only works in main thread

[Leuko]

Using a new process the following error occurs: cant pickle _thread.RLock objects. Any suggestions how I can solve this problem? Due to this post github.com/tiangolo/fastapi/issues/650 it is better to run it in a process but it's not working for me.

[Leuko]

Ok found a solution on my own. First it is important to use a new process to start uvicorn in it. Then you can kill or terminate the process if you want to stop uvicorn. But this does not seem to work on windows, at least for me it is just working on linux. To avoid the error of "cant pickle _thread.RLock objects" it is important not to use a method with self. So for example run_server(self) is not working with a new Process but run_server() is.

[Elijas Dapšauskas]

@Leuko Posted an answer with a proper fix to the "main thread" errors

[zzfima]

you are the king