Para usar una API de OAuth en mi aplicación local no web, envío al usuario al inicio de sesión de OAuth del servicio en un navegador, luego inicio temporalmente un servidor Flask para escuchar la solicitud de token de OAuth en localhost. En la ruta, llamo al shutdown para terminar el servidor temporal:
token = request.args["token"] shutdown = request.environ["werkzeug.server.shutdown"] shutdown() A partir de Flask 2.0 y Werkzeug 2.0, la función de shutdown del servidor de desarrollo está obsoleta. Al llamarlo, se muestra el siguiente mensaje:
La función 'environ['werkzeug.server.shutdown']' está obsoleta y se eliminará en Werkzeug 2.1.
Muchas respuestas de Stack Overflow sugieren werkzeug.server.shutdown , y ningún servidor WSGI de producción documenta cómo hacer esto. ¿Cómo puedo iniciar un servidor, esperar una sola solicitud, almacenar un valor y luego terminar el servidor?
Escribí una alternativa en la sección Apagar el servidor de los documentos de Werkzeug. Esto usa multiprocessing.Process para iniciar y esperar un proceso secundario, luego terminarlo una vez que se pasa un valor recibido a través de 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 Puede ver este trabajo agregando lo siguiente en la parte inferior y luego ejecutando el archivo con python :
if __name__ == "__main__": print(get_token()) Navegar a http://localhost:5000/?token=test imprimirá test y saldrá.
Otra alternativa similar que mostré en la discusión sobre la desaprobación es usar threading.Thread y make_server() en su lugar. Se puede demostrar de la misma manera que arriba.
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 Para Waitress , un servidor WSGI de producción que funcionará en Windows y Linux, el enfoque es casi idéntico. Reemplace run_simple() con waitress.serve() :
waitress.serve(app, host="localhost", port=5000) O para el enfoque de subprocesamiento, use waitress.create_server() y s.close() :
s = waitress.create_server(app, host="localhost", port=5000) t = threading.Thread(target=s.run) ... s.close() ...