如何关闭Flask服务器，实现OAuth回调获取code后自动退出

已尝试使用signals、keystroke logging、multiprocessing等方式关闭Flask服务器，均未成功。
预期实现逻辑：Flask应用启动运行后，从redirect_uri获取返回的授权code，随即自动退出服务器，继续执行主流程。
原有实现代码如下：

import webbrowser
from flask import Flask, request


app = Flask(__name__)

server = Process
code = ''
client_id = "4325234352342324"
redirect_uri = "http://localhost:80/shutdown"


@app.route('/')
def data():
    data = request.args
    return data

@app.after_request
def write_file():
    code = request.args['code']
    f = open('code.txt', 'w')
    f.write(code)

@app.before_first_request
def before_request():
    webbrowser.open(f"https://github.com/login/oauth/authorize?client_id={client_id}&redirect_uri={redirect_uri}&response_type=code")

@app.route('/shutdown')
def server_shutdown():
    func = request.environ.get('werkzeug.server.shutdown')
    if func is None:
        raise RuntimeError('Not running with the Werkzeug Server')
    func()

if __name__ == '__main__':
    app.run(host='localhost', port=80)

原有代码存在几个核心问题导致逻辑跑不通：

• @app.before_first_request装饰器在Flask 2.3及以上版本已被正式移除，无法触发自动打开授权页的逻辑
• @app.after_request装饰的函数必须接收响应对象作为参数、且执行完必须返回响应，原有写法会直接触发请求报错
• 80端口属于系统特权端口，非管理员/root权限下直接绑定会失败
• 依赖Werkzeug内置的shutdown方法兼容性差，部分版本中无法从request.environ中拿到该方法，导致关服务失败
• 全局变量在多进程场景下不互通，即使子进程拿到了code，主进程也无法直接读取

修正后可直接运行的代码 #

import time
import webbrowser
import multiprocessing
from flask import Flask, request

app = Flask(__name__)
client_id = "4325234352342324"
# 改用1024以上高位端口，无需管理员权限即可启动
redirect_uri = "http://localhost:8080/shutdown"

def run_server(queue: multiprocessing.Queue):
    auth_code = ""

    @app.route('/shutdown')
    def handle_auth_callback():
        nonlocal auth_code
        auth_code = request.args.get("code", "")
        if not auth_code:
            return "授权失败，未获取到有效code", 400
        return "授权成功，请返回终端继续操作"

    app.run(host="localhost", port=8080)
    # 服务退出时把拿到的code通过队列传回主进程
    queue.put(auth_code)

if __name__ == '__main__':
    # 用跨进程队列传递code，避免全局变量不生效
    code_queue = multiprocessing.Queue()
    # 把Flask服务跑在单独子进程
    server_process = multiprocessing.Process(target=run_server, args=(code_queue,))
    server_process.start()

    # 等待服务启动完成再拉起浏览器，避免连接被拒
    time.sleep(1)
    webbrowser.open(
        f"https://github.com/login/oauth/authorize?client_id={client_id}&redirect_uri={redirect_uri}&response_type=code"
    )

    # 轮询等待code返回
    received_code = ""
    while True:
        if not code_queue.empty():
            received_code = code_queue.get()
            break
        # 子进程异常退出时直接终止轮询
        if not server_process.is_alive():
            break
        time.sleep(0.5)

    # 拿到code后直接终止服务进程，无需依赖werkzeug的shutdown方法
    if server_process.is_alive():
        server_process.terminate()
        server_process.join()

    # 以下写后续主流程逻辑，比如用code换access_token等
    print(f"成功获取授权code：{received_code}")

关键注意点 #

• 不再依赖Werkzeug内置的shutdown接口，直接通过终止子进程的方式关闭服务，所有Flask、Werkzeug版本都能兼容
• 授权回调逻辑统一放在/shutdown路由下处理，去掉了逻辑有问题的after_request、废弃的before_first_request装饰器
• 用multiprocessing.Queue做跨进程通信，保证主进程能稳定拿到子进程获取到的code
• 加了服务启动等待时间，避免浏览器提前打开导致连接失败
• 轮询逻辑加了子进程存活检测，避免服务异常时出现死循环

内容的提问来源于stack exchange，提问作者pt_dev