Uvicorn cannot be shutdown programmatically

[Andrew-lll]

There is no documented way to shutdown uvicorn in python:

ex:

instance = uvicorn.run("example:app", host="127.0.0.1", port=5000, log_level="info")
instance.shutdown()

How do we shutdown uvicorn?

[florimondmanca]

Hi,

Not documented indeed, but a multithreaded approach should do…

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 started.
    ...
# 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

[Robby-Sodhi]

When I try to run the above example with a number of workers in the config, they don't start (seems like that flag is just ignored?)

config:

config = uvicorn.Config("rest:app", host="192.168.0.23", port=5000, reload=False, workers=4)

output:

←[32mINFO←[0m:     Started server process [←[36m25028←[0m]
←[32mINFO←[0m:     Waiting for application startup.
←[32mINFO←[0m:     Application startup complete.
←[32mINFO←[0m:     Uvicorn running on ←[1mhttp:https://192.168.0.23:5000←[0m (Press CTRL+C to quit)

expected:

←[32mINFO←[0m:     Uvicorn running on ←[1mhttp:https://192.168.0.23:5000←[0m (Press CTRL+C to quit)
←[32mINFO←[0m:     Started parent process [←[36m←[1m26384←[0m]
←[32mINFO←[0m:     Started server process [←[36m9396←[0m]
←[32mINFO←[0m:     Waiting for application startup.
←[32mINFO←[0m:     Started server process [←[36m33344←[0m]
←[32mINFO←[0m:     Waiting for application startup.
←[32mINFO←[0m:     Application startup complete.
←[32mINFO←[0m:     Started server process [←[36m25068←[0m]
←[32mINFO←[0m:     Application startup complete.
←[32mINFO←[0m:     Waiting for application startup.
←[32mINFO←[0m:     Application startup complete.
←[32mINFO←[0m:     Started server process [←[36m20548←[0m]
←[32mINFO←[0m:     Waiting for application startup.
←[32mINFO←[0m:     Application startup complete.

Expected output is using:

uvicorn.run("rest:app", host="192.168.0.23", port=5000, reload=False, workers=4)

[Andrew-lll]

I do appreciate the reply/code, but:

Why is this not built in and documented?
Why do I have to implement this code?

It's a webserver. Run_in_thread() and shutdown() are core functionality...am I wrong?

[plazmakeks]

I'm following @florimondmanca s approach to run pact tests on the provider side. However when using this i get the following error from uvicorn's Server.run method:

RuntimeError: There is no current event loop in thread 'Thread-7'.

Any idea on that? Happens with python 3.6.9 as well as 3.7.7.

another edit:

I got it to work using

config = Config("example:app", host="127.0.0.1", port=5000, log_level="info", loop="asyncio")

[Elijas]

I do appreciate the reply/code, but:

Why is this not built in and documented?
Why do I have to implement this code?

It's a webserver. Run_in_thread() and shutdown() are core functionality...am I wrong?

Fully agreed, this question of how to run and stop unicorn pops up on stack overflow as well, it's quite obscure as it is now

[florimondmanca]

I appreciate the feedback here, but then this brings the question — what would folks expect the usage API to be for something like this?

I think a nice approach for allowing a programmatic shutdown of Uvicorn would be to exit the space of .run() and embrace concurrency, that is the async serve() method. It's actually super easy to spin up a concurrent task, and then clean it up whenever the server isn't needed anymore. (We could also very well make it easier to do this multithreaded thing, and having with server.run_in_thread() built in, or something.)

But then there's the question of A/ waiting for the server to startup (without having to reinvent the wheel each time), and B/ triggering the server shutdown and have it clean up gracefully (straight up cancelling a task is very... brutal).

Personally, I'd love this kind of API:

async with open_task_group() as tg:
    tg.start_soon(server.serve)
    await server.wait_started()
    # ...
    # Cause serve() to terminate soon, allowing the task
    # to finish cleanly when exiting the context manager.
    server.shutdown_soon().

Here open_task_group() refers to some kind of trio-like nursery API. On asyncio this could be a simplified helper, like this:

@asynccontextmanager
async def start_concurrently(async_fn):
    task = asyncio.create_task(async_fn())
    try:
        yield
    finally:
        await task

It could be used like this:

async with start_concurrently(server.serve):
    await server.wait_started()
    # ...
    server.shutdown_soon()

I'm also happy to discuss a threaded equivalent API, for use cases where an event loop isn't readily available or practical (for example, when testing the server using a sync HTTP client).

Then we need to figure out how to make these APIs come to life. :-) Maybe this means we'll need sync/async equivalents, like .wait_started() vs .await_started(), etc.

[Elijas]

That's quite an elaborate pondering, thank you for that! I'd be willing to use all these APIs:)

Just for curiosity

If I understand correctly, having the server.shutdown_soon() implicit in the context itself like in your original example would be considered confusing? At least for my simple use cases, wait_started would immediatelly follow start and context teardown would always implicitly include shutdown_soon anyway - if it's not a part of the lib, I'd write a facade/subclass for it - I think it's simpler to just have something along the lines of

with running_server():
    ... # Make requests to the server, etc.

[florimondmanca]

Well, yes, if we add a .serve_concurrently() method that does the "wait for started on enter" and "shutdown soon then await tasks on exit" thing, that's also an option. Probably also more convenient to use than having to deal with library concurrency APIs. (Let's face it, unlike trio asyncio doesn't offer safe enough APIs that would allow us to tell people to rely on them.)

And then a .run_in_thread() equivalent for threaded contexts.

Now that we know this is the kind of thing we want to have eventually, anyone that'd like to figure out internals and ways to implement this is very much welcome to. :) I'll reopen since I think this is indeed a valuable thing to add.

[roboto84]

For people that stop by here and want to try out florimondmanca's multithreaded approach and are a bit new to uvicorn, Config in the code comes from uvicorn. Configuration below also includes plazmakeks' addition in the case of running into a runtime error described by them.

config = uvicorn.Config(app, host="127.0.0.1", port=5000, log_level="info", loop="asyncio")

Thanks for this discussion all.

[jlaw]

For people that stop by here and want to try out florimondmanca's multithreaded approach and are a bit new to uvicorn, Config in the code comes from uvicorn. Configuration below also includes plazmakeks' addition in the case of running into a runtime error described by them.

config = uvicorn.Config(app, host="127.0.0.1", port=5000, log_level="info", loop="asyncio")

Thanks for this discussion all.

I also encountered this RuntimeError and I did not want to force loop="asyncio", so I implemented this override to run():
freqtrade/freqtrade#4530

I think it would help everyone if the following code was implemented for https://github.com/encode/uvicorn/blob/master/uvicorn/loops/uvloop.py:

  import asyncio
+ import threading

  import uvloop


  def uvloop_setup():
-     asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+     if (threading.current_thread() is threading.main_thread()):
+         asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+     else:
+         asyncio.set_event_loop(uvloop.new_event_loop())

[rams3sh]

@florimondmanca I used your code and found the code to be ever running in case self.started was never set. In my case , a port was already in use. But the program does not stop on exception here. Added some output Log below :-

2021-04-08 11:54:41,881 — [Module Name: uvicorn.error] — [PID: 25125] — [Thread : Thread-1] —  INFO — [Method and Line No: serve:64] — Started server process [25125]
INFO:     Waiting for application startup.
2021-04-08 11:54:41,881 — [Module Name: uvicorn.error] — [PID: 25125] — [Thread : Thread-1] —  INFO — [Method and Line No: startup:26] — Waiting for application startup.
INFO:     Application startup complete.
2021-04-08 11:54:41,881 — [Module Name: uvicorn.error] — [PID: 25125] — [Thread : Thread-1] —  INFO — [Method and Line No: startup:38] — Application startup complete.
ERROR:    [Errno 98] error while attempting to bind on address ('127.0.0.1', 8000): address already in use
2021-04-08 11:54:41,881 — [Module Name: uvicorn.error] — [PID: 25125] — [Thread : Thread-1] —  ERROR — [Method and Line No: startup:150] — [Errno 98] error while attempting to bind on address ('127.0.0.1', 8000): address already in use
INFO:     Waiting for application shutdown.
2021-04-08 11:54:41,881 — [Module Name: uvicorn.error] — [PID: 25125] — [Thread : Thread-1] —  INFO — [Method and Line No: shutdown:43] — Waiting for application shutdown.
INFO:     Application shutdown complete.
2021-04-08 11:54:41,882 — [Module Name: uvicorn.error] — [PID: 25125] — [Thread : Thread-1] —  INFO — [Method and Line No: shutdown:46] — Application shutdown complete.
2021-04-08 11:54:42,882 — [Module Name: helper.status ] — [PID: 25125] — [Thread : Thread-2] —  INFO — [Method and Line No: helper:23] — Running ...
2021-04-08 11:54:42,882 — [Module Name: helper.status ] — [PID: 25125] — [Thread : Thread-2] —  INFO — [Method and Line No: helper:23] — Running ...

And this line was never met. Hence I have modified the code as below . Since I am not familiar with the internals of uvicorn server, I may not know if the below code has some edge case or anything that I am missing. Let me know if I am missing anything !!

class Server(uvicorn.Server):

    def install_signal_handlers(self):
        pass

    @contextlib.contextmanager
    def running(self):
        thread = threading.Thread(target=self.run)
        thread.start()
        try:
            while not self.started and thread.is_alive(): # Added a condition for checking if thread is alive 
                time.sleep(1e-3)
            yield
        finally:
            self.should_exit = True
            thread.join()

[Pharisaeus]

I am very confused why this requires some crazy-hoops. I checked the uvicorn.run() code and I don't understand why can't this logic be split into 2 parts:

def run(app, **kwargs):
    runner = create_runner(app, kwargs)
    runner.run()

def create_runner(app, **kwargs):
    config = Config(app, **kwargs)
    server = Server(config=config)

    if (config.reload or config.workers > 1) and not isinstance(app, str):
        logger = logging.getLogger("uvicorn.error")
        logger.warning(
            "You must pass the application as an import string to enable 'reload' or "
            "'workers'."
        )
        sys.exit(1)

    if config.should_reload:
        sock = config.bind_socket()
        supervisor = ChangeReload(config, target=server.run, sockets=[sock])
        return supervisor
    elif config.workers > 1:
        sock = config.bind_socket()
        supervisor = Multiprocess(config, target=server.run, sockets=[sock])
        return supervisor
    else:
        return server

This way someone can still do uvicorn.run() with exactly the same behaviour, but someone else can also manually call runner = uvicorn.create_runner() and then do runner.run(), and have the ability to set runner.should_exit flag.
One comment from me that it would be nice if those 3 types you can get back (Server, ChangeReload and Multiprocess) had some common interface to do this like some close() method. Right now the should_exit flag is either Event or boolean, depending on scenario, and we would have to check the type before doing .set() or = True and this should be delegated to the appropriate class.

[rmorshea]

Anyone care to review #1011? It doesn't solve this issue completely, but it would make the solution a little simpler.

[MauricePasternak]

I have not had success in triggering this to shut down from within a FastAPI route. Could anyone shed light as to how this can be made to work without resorting to psutil process-killing?

import contextlib
import time
import threading
import uvicorn
from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def greet():
    return {"message": "Hello, world"}

@app.get("/kill")
async def kill():
    global server
    server.keep_running = False

class Server(uvicorn.Server):
    def __init__(self, config):
        super().__init__(config)
        self.keep_running = True

    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
            while self.keep_running:
                time.sleep(1e-3)
        finally:
            self.should_exit = True
            thread.join()

config = uvicorn.Config("main:app", host="127.0.0.1", port=5000, log_level="info")
server = Server(config=config)

if __name__ == "__main__":
    with server.run_in_thread():
        while server.keep_running:
            pass

[euri10]

migrating this as a discussion following this comment from @rmorshea

let's come up with a design and we see if we add it or not, I still dont get why subclassing the server is not sufficient, but I admit I didn't spend much time on this, nor have the will to do it, so someone will have to step up and eli5 me or other maintainers the ins and outs.

just for the record, the whole test suite was using this run in a thread pattern and it was just ugly, flaky, full of undesirable and different side-effects on the various platforms we support, so we removed it and so far the experience is way better without it.

[Aponace]

I would like to suggest an easier solution:

import multiprocessing
from time import sleep

from fastapi import FastAPI
from uvicorn import Config, Server


class UvicornServer(multiprocessing.Process):

    def __init__(self, config: Config):
        super().__init__()

        self.config = config

    def stop(self):
        self.terminate()

    def run(self, *args, **kwargs):
        server = Server(config=self.config)
        server.run()


app = FastAPI()


def main():  # POC
    u = UvicornServer(config=Config(app=app))

    u.start()
    sleep(5)  # Start timeout

    u.stop()
    sleep(5)  # Stop timeout & Prevent EOF termination


if __name__ == '__main__':
    main()

[ruijiexie]

I also encountered the same problem. Here is my solution

import time
import uvicorn
import asyncio
import sys
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()

@app.get("/kill")
async def kill():
    global server
    global loop
    loop.stop()
    await server.shutdown()

config = uvicorn.Config(app=app, host="127.0.0.1", port=5000, log_level="debug")
server = uvicorn.Server(config=config)
loop = None

if __name__ == "__main__":
	try:
		config.setup_event_loop()
		loop = asyncio.get_event_loop()
		loop.run_until_complete(server.serve())
	except:
		print(sys.exc_info())
	while True:
		time.sleep(1)

[pazaan]

In a similar vein to @ruijiexie, but a little simpler (includes capturing coverage):

import uvicorn
import atexit
import coverage

cov = coverage.Coverage()
cov.start()

# pylint: disable=wrong-import-position
from fastapi import FastAPI

app = FastAPI()


def save_coverage():
    print("saving coverage")
    cov.stop()
    cov.save()


@app.get("/shutdown")
async def shutdown():
    global server
    server.should_exit = True
    server.force_exit = True
    await server.shutdown()


atexit.register(save_coverage)

config = uvicorn.Config(app, host="0.0.0.0", port=8000, log_level="info", loop="asyncio")
server = uvicorn.Server(config=config)
server.run()

[gh-ppolk]

Hey, I am running into exactly this setup issue using the same tools - pytest, fastapi, and uvicorn. Another work around I have seen is to use docker containers instead of process threads and just kill the docker containers. Wanted to add that to the conversation.

[arthur-tacca]

I don't see why you need to inherit from uvicorn.Server to set server.should_exit = True. It doesn't have a leading underscore so presumably it's meant to be public, so you can just set it on any Server instance. (If it's not meant to be a public member then it ought to be called server._should_exit by normal Python convention.)

In any case, there's a more normal way to cancel async tasks: async task cancellation. Just run await server.serve() in a task and cancel that task when desired, then await server.shutdown(). It's a bit odd that cancelling server.serve() doesn't just call server.shutdown() automatically; probably that should be a bug in its own right.

async def server_run(server):
    try:
        await server.serve()
    finally:
        await server.shutdown()

[Edit: I just noticed that the original request doesn't even mention threads. If you just want to run a uvicorn server in a separate asyncio task and then stop it later, you can just use asyncio.create_task(server_run(server)) with the above function. When you want to shut it down, cancel the task, then await it to find out when cancellation completes (and ignore the CancelledError that it throws).]

Cancellation of asyncio tasks from other threads is a bit of faff but that's more asyncio's fault than uvicorn's. You need to create a task and capture the asyncio loop so you can call the task's cancel method in the other thread. For example, it could look like this (assuming this is part of a wider class but I've not bothered to fill it all in):

async def run(self, server):
    self.loop = asyncio.get_running_loop()
    self.task = asyncio.create_task(_server_run(server))
    try:
        await self.task
    except asyncio.CancelledError:
        pass
    finally:
        await server.shutdown()

You would stop the server like this:

def cancel(self):
     self.loop.call_soon_threadsafe(self.task.cancel)

Assuming you start a thread with threading.Thread(target=asyncio.run, args=(self.run(server),)), you can then call thread.join() to wait for it to finish.

Gracefully waiting for startup to complete, if that's important, is harder. It's a pity that the best suggestion is to loop waiting for server.started. Better would be if you could call server.startup() and server.main_loop() separately, so that you could signal threading.Event in between. But it appears that server.serve() does some other admin before calling server.startup() so I don't think would work. (And therefore startup() seems to be another case where something ought to be private but doesn't have a leading underscore.)

[WolfgangFahl]

Could you please provide a complete working example? I could not get this working

[meoyawn]

some good suggestions here https://stackoverflow.com/q/57412825/1032286

[WolfgangFahl]

To bad his is in discussion mode and not handled as an issue any more. It would be great to hide the complexity behind this especially since the multiprocessing suggestion doesn't cut it in case there is non-pickable stuff.

[whatamithinking]

Apologies if i am missing the obvious, but is there a reason we cannot add a should_exit_timeout to the config and then update the loop in shutdown to check it has not exceeded this deadline?

For example, I plan on using (i.e. have not tested) the following to make this work.

class Config(uvicorn.Config):

    def __init__(self, should_exit_timeout: Optional[float] = None, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        self.should_exit_timeout = should_exit_timeout


class Server(uvicorn.Server):
    
    async def shutdown(self, sockets: Optional[List[socket.socket]] = None) -> None:
        logger = logging.getLogger("uvicorn.error")
        logger.info("Shutting down")
        
        should_exit_deadline = None
        if self.config.should_exit_timeout is not None:
            should_exit_deadline = time.time() + self.config.should_exit_timeout

        # Stop accepting new connections.
        for server in self.servers:
            server.close()
        for sock in sockets or []:
            sock.close()
        for server in self.servers:
            await server.wait_closed()

        # Request shutdown on all existing connections.
        for connection in list(self.server_state.connections):
            connection.shutdown()
        await asyncio.sleep(0.1)
        
        # Wait for existing connections to finish sending responses.
        if self.server_state.connections and not self.force_exit:
            msg = "Waiting for connections to close. (CTRL+C to force quit)"
            logger.info(msg)
            while self.server_state.connections and not self.force_exit and \
                (should_exit_deadline is None or time.time() < should_exit_deadline):
                await asyncio.sleep(0.1)

        # Wait for existing tasks to complete.
        if self.server_state.tasks and not self.force_exit:
            msg = "Waiting for background tasks to complete. (CTRL+C to force quit)"
            uvicorn.logger.info(msg)
            while self.server_state.tasks and not self.force_exit:
                await asyncio.sleep(0.1)

        # Send the lifespan shutdown event, and wait for application shutdown.
        if not self.force_exit:
            await self.lifespan.shutdown()


uvicorn.Server.shutdown = shutdown

[igz-gp]

I may have missed something, but it's really easy to do this extending the server as @euri10 said previously

Naive example:

class UvicornServer(uvicorn.Server):
    serve_task: Task
    did_start: Event
    did_close: Event

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.did_start = Event()
        self.did_close = Event()

    async def start(self):
        self.serve_task = asyncio.create_task(self.serve())
        self.serve_task.add_done_callback(lambda _: self.did_close.set())
        await self.did_start.wait()

    async def startup(self, sockets: list = None) -> None:
        await super().startup(sockets)
        self.did_start.set()

    async def shutdown(self):
        await super().shutdown()
        self.serve_task.cancel()
        await self.did_close.wait()

[whatamithinking]

@igz-gp If you review the code for Server.shutdown, you will see that by default it politely waits for all connections to be closed. If you have SSE/long-polling endpoints, the server will hang forever on the call to Server.shutdown.

I could be wrong, but it does not look like shutdown is setup to handle cancellation itself. If you try to run shutdown in a task and cancel it or use the new asyncio.timeout context manager, you may interrupt a critical part of the shutdown process. We usually just want to give up on waiting for the parts of shutdown which can hang forever and continue with the rest of shutdown after that.

Other solutions might be:

• shield the critical sections from cancellation in Server.shutdown and timeout the task
• keep a list of all connections which are for blocking requests and have Server.shutdown exit when only those connections remain
  • this may be tough in practice as you may not know something will block for a long time in advance
• add asyncio.Event primitives in the requests which will hang forever and check them periodically and exit if they are set
  • similar problem to the option above. the section of code which blocks could be deep in the stack, which would probably make this a non-option in some cases.

[igz-gp]

@ConnorSMaynes IMHO, I personally would use asyncio.timeout or asyncio.wait_for the server.shutdown().
(I mean... If a sysadmin asks a server to shutdown and the server turns unresponsive, what do you think he will do?)
This should obviously be tuned for each case. In my case, I don't even need to apply a timeout as I know there won't be any long-polling endpoint. Don't know how it should work for long-lived connections or web sockets, though. But again, that's a different use-case.

[Gallagher-Polyn-DCO]

python3 -m uvicorn app.main:app --reload --workers 1 --host 0.0.0.0 --port 8000

from fastapi import FastAPI
import os

app = FastAPI()

@app.get("/chaos/kill")
async def kill_uvicorn():
    parent_pid = os.getppid()
    os.kill(parent_pid, 9)

[particle1331]

This one worked for testing out restart behaviors of my container. Short and sweet. :)

[Gallagher-Polyn-DCO]

One thing I later discovered, another process, I guess the child process, stays open on the port -- so one might want to kill that process, too.

[rodja]

Please note that getppid is not available on Windows. This may not be of concern to most users, but for our web-frontend library NiceGUI we are still seeking for a true cross-platform solution.

[WolfgangFahl]

same for justpy

[arifikhsan]

cleanest solution so far

[Emptyless]

If you are fine with running Uvicorn in a separate thread, it is possible to start/stop without relying on async:

class ThreadedUvicorn:
    def __init__(self, config: Config):
        self.server = uvicorn.Server(config)
        self.thread = threading.Thread(daemon=True, target=self.server.run)

    def start(self):
        self.thread.start()
        asyncio.run(self.wait_for_started())

    async def wait_for_started(self):
        while not self.server.started:
            await asyncio.sleep(0.1)

    def stop(self):
        if self.thread.is_alive():
            self.server.should_exit = True
            while self.thread.is_alive():
                continue

server = ThreadedUvicorn(app, host=host, port=port)
server.start()
server.stop()

@rodja this should also work cross-platform. Might need to check if the platform is win32 and run asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) for the asyncio part.

[acode-x]

With the help of above code, I am able to start and stop the server. However on restart I get error:

ERROR:    Traceback (most recent call last):
  File "/home/scc-dev/edge_/env/lib/python3.8/site-packages/starlette/routing.py", line 686, in lifespan
    await receive()
  File "/home/scc-dev/edge_/env/lib/python3.8/site-packages/uvicorn/lifespan/on.py", line 137, in receive
    return await self.receive_queue.get()
  File "/usr/lib/python3.8/asyncio/queues.py", line 163, in get
    await getter
asyncio.exceptions.CancelledError

am using the same function, with a small change where I initialise thread within start function.
Any thoughts on same?

[826541814]

it's working for me, thank you

restart service only need to run this

server = ThreadedUvicorn(app, host=host, port=port)
server.start()
server.stop()
# restart 
server = ThreadedUvicorn(app, host=host, port=port)
server.start()

[acode-x]

Thanks! I just found more workaround

    @contextmanager
    def run_in_thread(self):
        if self.should_exit:
            self.should_exit = False
        thread = threading.Thread(target=self.run)
        thread.start()
        try:
            while not self.started:
                sleep(1e-3)
            yield
        finally:
            self.should_exit = True
            thread.join()
            logging.debug("Server stopped")

[BerengarWLehr]

Okey, I have a one-line solution, please provide explenation why this is also a bad idea:

_thread.interrupt_main()

(the need to import _thread is implied)

[WolfgangFahl]

see also zauberzeug/nicegui#1956

[Booplicate]

I'm surprised nobody suggested something like

signal.raise_signal(signal.SIGINT)

If you're running in the main thread, then uvicorn already has signal handlers set up for you, so all we need is to emulate a signal to shutdown.

[dorinclisu]

Virtually the same thing was already suggested by #1103 (comment)

But even that comes late to the discussion, I guess many programmers like to complicate things..

[arthur-tacca]

@dorinclisu Did you link to the wrong thing by mistake? Because that solution (kill -9, i.e. SIGKILL, to parent process) is nothing like this one (SIGINT to current progress).

[dorinclisu]

Surely using only the signal library to raise the signal is more elegant, but actually os.getpid() gets current process not parent process, and then yeah just replace SIGKILL with SIGTERM/SIGINT, the idea is the same.

[arthur-tacca]

That answer is using os.getppid(), not os.getpid(). They're using it in a situation where there is a separate worker process (that's what the command line at the start does), so the Python runs in that worker, which means the parent is the main server process.