Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

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

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

like image 367
Leuko Avatar asked May 03 '20 16:05

Leuko


People also ask

Is FastAPI single threaded?

Amazing single-threaded performance with FastAPI — optimize your code for a HUGE performance boost! Python has many web frameworks the most popular being Django and Flask.

Do I need Gunicorn with Uvicorn?

Running with GunicornFor production deployments we recommend using gunicorn with the uvicorn worker class. For a PyPy compatible configuration use uvicorn.

How do you stop Uvicorn?

One way is to run it, and then to type ^Z to pause it. And then type "bg" to continue it in the background. And then type "exit" to the shell to make the shell and terminal go away. Alternatively, you could initially run it with "&" on the end of the command line, and elide the ^Z and the "bg".


1 Answers

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

like image 99
Elijas Dapšauskas Avatar answered Sep 20 '22 18:09

Elijas Dapšauskas