如何在10秒后正常关闭Flask应用（无需Ctrl+C）并解决报错

问题描述 #

我想把Flask作为Python脚本的一部分使用：先执行一些命令，启动Flask服务器来发送文件，10秒或更久后关闭服务器，继续执行脚本。但执行shutdown_server函数时出现报错，必须按下Ctrl+C才能让脚本继续运行。

原代码 #

from flask import send_from_directory, abort 
from flask import Flask
import socket
from flask import request
from flask_apscheduler import APScheduler


hostName= socket.gethostname()
ipAddr= socket.gethostbyname(hostName)
app= Flask(__name__)
app.config["CLIENT_IMAGES"]="photo"

socketio = SocketIO(app)
sched=APScheduler()
counter=0

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

@app.route("/")
def index():
    return "Hello World"


@app.route("/get-image/<string:image_name>") 
def get_image(image_name):    
    try:
        return send_from_directory(app.config["CLIENT_IMAGES"], "photo.jpg", as_attachment=True)
        
    except FileNotFoundError:
        abort(404)

if __name__=="__main__":
    sched.add_job(id='Job1',func=shutdown_server, trigger='interval', seconds=5)
    sched.start()
    #app.run()
    #socketio.run(app,host="0.0.0.0", port=4998,allow_unsafe_werkzeug=True)
    app.run(host=ipAddr, port='5000',debug= False) #,use_reloader=False 
    #os.system("start /b python app.py")
    #os.system("python "+ "app.py &" )
counter+=1
print("ALohaaa")

报错信息 #

Job "Job1 (trigger: interval[0:00:05], next run at: 2023-10-09 09:59:57 EEST)" 抛出异常
Traceback (most recent call last):
  File "C:\Users\userName\anaconda3\Lib\site-packages\apscheduler\executors\base.py", line 125, in run_job
    retval = job.func(*job.args, **job.kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "c:\users\ userName \desktop\jetson\camera test\mix\qr prog test\try03.py", line 21, in shutdown_server
    func = request.environ.get('werkzeug.server.shutdown')
           ^^^^^^^^^^^^^^^
  File "C:\Users\ userName \anaconda3\Lib\site-packages\werkzeug\local.py", line 311, in __get__
    obj = instance._get_current_object()
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\ userName \anaconda3\Lib\site-packages\werkzeug\local.py", line 508, in _get_current_object
    raise RuntimeError(unbound_message) from None
RuntimeError: 在请求上下文之外工作。

这通常意味着你尝试使用需要活跃HTTP请求的功能。请查阅测试相关文档了解如何避免此问题。

问题原因 #

shutdown_server函数中调用了request对象，但APScheduler的定时任务是在独立线程中执行的，没有绑定任何HTTP请求上下文，因此无法获取request对象，导致报错。另外app.run()是阻塞式调用，即便定时任务不报错，也需要服务器关闭后主线程才会继续执行后续代码。

解决方案 #

方案1：用线程+等待替代APScheduler（推荐） #

直接通过线程启动Flask服务器，主线程等待指定时间后关闭服务器，无需依赖APScheduler，逻辑更简单：

from flask import send_from_directory, abort 
from flask import Flask
import socket
import threading
import time
from werkzeug.serving import run_simple

hostName = socket.gethostname()
ipAddr = socket.gethostbyname(hostName)
app = Flask(__name__)
app.config["CLIENT_IMAGES"] = "photo"

server = None
counter = 0

@app.route("/")
def index():
    return "Hello World"

@app.route("/get-image/<string:image_name>") 
def get_image(image_name):    
    try:
        return send_from_directory(app.config["CLIENT_IMAGES"], "photo.jpg", as_attachment=True)
    except FileNotFoundError:
        abort(404)

def start_server():
    global server
    # 用run_simple启动服务器，保存实例以便后续关闭
    server = run_simple(ipAddr, 5000, app, use_reloader=False)

def shutdown_server():
    if server:
        server.shutdown()

if __name__ == "__main__":
    # 启动服务器线程
    server_thread = threading.Thread(target=start_server)
    server_thread.start()
    
    # 等待10秒，可根据需求调整时间
    time.sleep(10)
    
    # 关闭服务器并等待线程结束
    shutdown_server()
    server_thread.join()
    
    # 继续执行后续代码
    counter += 1
    print("ALohaaa")

方案2：保留APScheduler，移除请求上下文依赖 #

如果必须使用APScheduler，修改shutdown_server函数，直接操作服务器实例，避免使用request对象：

from flask import send_from_directory, abort 
from flask import Flask
import socket
from flask_apscheduler import APScheduler
from werkzeug.serving import run_simple

hostName = socket.gethostname()
ipAddr = socket.gethostbyname(hostName)
app = Flask(__name__)
app.config["CLIENT_IMAGES"] = "photo"

# 启用APScheduler API（可选）
app.config['SCHEDULER_API_ENABLED'] = True
sched = APScheduler(app)
server = None
counter = 0

def shutdown_server():
    global server
    if server:
        server.shutdown()

@app.route("/")
def index():
    return "Hello World"

@app.route("/get-image/<string:image_name>") 
def get_image(image_name):    
    try:
        return send_from_directory(app.config["CLIENT_IMAGES"], "photo.jpg", as_attachment=True)
    except FileNotFoundError:
        abort(404)

if __name__ == "__main__":
    # 添加定时关闭任务，延迟10秒执行
    sched.add_job(id='Job1', func=shutdown_server, trigger='interval', seconds=10)
    sched.start()
    
    # 启动服务器并保存实例
    global server
    server = run_simple(ipAddr, 5000, app, use_reloader=False)
    
    # 服务器关闭后执行后续代码
    counter += 1
    print("ALohaaa")

说明 #

• 两种方案都绕过了请求上下文的限制，直接通过服务器实例调用shutdown()方法关闭服务器。
• use_reloader=False必须设置，否则会启动两个服务器实例，导致关闭失效。

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