wait for value then stop server, after 'werkzeug.server.shutdown' is deprecated and removed

In order to use an OAuth API in my local, non-web application, I send the user to the service's OAuth login in a browser, then temporarily start a Flask server to listen for the OAuth token request on localhost. In the route, I call shutdown to terminate the temporary server:

token = request.args["token"]
shutdown = request.environ["werkzeug.server.shutdown"]
shutdown()

As of Flask 2.0 and Werkzeug 2.0, the dev server shutdown function is deprecated. When calling it, the following message is shown:

The 'environ['werkzeug.server.shutdown']' function is deprecated and will be removed in Werkzeug 2.1.

Many Stack Overflow answers suggest werkzeug.server.shutdown, and no production WSGI servers document how do this. How can I start a server, wait for a single request, store a value, then terminate the server?

Solution

I wrote an alternative in the Shutting Down the Server section of Werkzeug's docs. This uses multiprocessing.Process to start and wait for a child process, then terminate it once a received value is passed back over a multiprocessing.Queue.

import multiprocessing
from werkzeug import Request, Response, run_simple

def run_token_server(q: multiprocessing.Queue) -> None:
    @Request.application
    def app(request: Request) -> Response:
        q.put(request.args["token"])
        return Response("", 204)

    run_simple("localhost", 5000, app)


def get_token():
    q = multiprocessing.Queue()
    p = multiprocessing.Process(target=run_token_server, args=(q,))
    p.start()
    token = q.get(block=True)
    p.terminate()
    return token

You can see this work by adding the following to the bottom, then running the file with python:

if __name__ == "__main__":
    print(get_token())

Navigating to http://localhost:5000/?token=test will print test and exit.

Another similar alternative I showed in the deprecation discussion is to use threading.Thread and make_server() instead. It can be demonstrated the same way as above.

import threading
from queue import Queue

from werkzeug import Request, Response
from werkzeug.serving import make_server


def get_token():
    @Request.application
    def app(request):
        q.put(request.args["token"])
        return Response("", 204)
    
    q = Queue()
    s = make_server("localhost", 5000, app)
    t = threading.Thread(target=s.serve_forever)
    t.start()
    token = q.get(block=True)
    s.shutdown()
    t.join()
    return token

For Waitress, a production WSGI server that will work on Windows and Linux, the approach is almost identical. Replace run_simple() with waitress.serve():

waitress.serve(app, host="localhost", port=5000)

Or for the threading approach use waitress.create_server() and s.close():

s = waitress.create_server(app, host="localhost", port=5000)
t = threading.Thread(target=s.run)
...
s.close()
...