Search code examples
pythonmultiprocessingfastapiuvicorn

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?


Solution

  • 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 :


    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 :