Flask服务与外部Python脚本共享可变变量问题求助

Flask服务和你的外部脚本是两个独立的Python进程，各自拥有独立的内存空间。直接通过from app import user导入的只是脚本启动时的实例副本，Flask进程中对实例的修改不会同步到外部脚本的内存里，自然拿不到最新数据。

方案1：用缓存/数据库实现进程间数据共享（推荐） #

通过第三方存储（比如Redis、SQLite）作为数据中转，Flask把更新后的数据存进去，外部脚本从存储中读取最新值。这里以Redis为例：

步骤1：安装依赖 #

pip install redis

修改Flask应用（app.py） #

from flask import Flask, request, render_template
import redis

app = Flask(__name__)
# 连接本地Redis实例（默认端口6379）
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

@app.route('/', methods=["GET", "POST"])
@app.route('/index', methods=["GET", "POST"])
def index():
    if request.method == "POST":
        data = request.get_json()
        if "something" in data:
            # 将新值存入Redis
            redis_client.set("user_something", data["something"])
            # 如果需要主动通知外部脚本，可发布消息
            redis_client.publish("something_updates", data["something"])
    return render_template('index.html')

if __name__ == '__main__':
    app.run()

外部脚本代码 #

import redis

redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# 方式1：主动轮询获取最新值
print("当前值:", redis_client.get("user_something"))

# 方式2：订阅Redis消息，实时接收变更（无需轮询）
pubsub = redis_client.pubsub()
pubsub.subscribe("something_updates")
print("等待接收数据变更...")
for message in pubsub.listen():
    if message['type'] == 'message':
        print("收到新值:", message['data'])

方案2：给Flask加接口，外部脚本通过HTTP请求获取数据 #

在Flask中新增一个接口专门返回something的最新值，外部脚本通过请求该接口拿到数据。

修改Flask应用（app.py） #

from flask import Flask, request, render_template, jsonify

app = Flask(__name__)

class User:
    def __init__(self):
        self._something = 0

    @property
    def something(self):
        return self._something

    @something.setter
    def something(self, value):
        self._something = value

user = User()

@app.route('/', methods=["GET", "POST"])
@app.route('/index', methods=["GET", "POST"])
def index():
    if request.method == "POST":
        data = request.get_json()
        if "something" in data:
            user.something = data["something"]
    return render_template('index.html')

# 新增获取数据的接口
@app.route('/get-something', methods=["GET"])
def get_something():
    return jsonify({"something": user.something})

if __name__ == '__main__':
    app.run()

外部脚本代码 #

import requests

# 请求Flask接口获取最新值
response = requests.get("http://localhost:5000/get-something")
if response.status_code == 200:
    data = response.json()
    print("当前值:", data["something"])

方案3：使用进程共享内存（局限性大） #

如果仅在单机器、单进程Flask场景下使用，可以用Python的multiprocessing模块实现共享内存，但Flask若以多进程模式启动（比如Gunicorn多进程），该方案会失效。

修改User类（user.py） #

from multiprocessing import Value

class User:
    def __init__(self):
        # 使用共享整数存储值（'i'表示整数类型）
        self._something = Value('i', 0)

    @property
    def something(self):
        # 加锁避免并发读写冲突
        with self._something.get_lock():
            return self._something.value

    @something.setter
    def something(self, value):
        with self._something.get_lock():
            self._something.value = value

外部脚本代码 #

from app import user

print("当前值:", user.something)

• 推荐用方案1（Redis缓存）：支持实时推送、高性能，适合大多数生产场景。
• 简单需求可选方案2（HTTP接口）：实现成本低，无需额外依赖。
• 方案3仅适合本地单进程测试场景，不推荐生产使用。

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