如何在运行的 Flask 实例中重新加载配置数据？

Question

假设我们有一个使用文件系统配置文件来完成某些工作的服务：

class Service:

    def _load():
        # (re)loads config file
        pass

    def do_some_work():
        # uses config file
        pass  

...以及一个为其客户提供服务的 Flask 应用程序。

app = Flask(__name__)
service = Service()

@app.route('/do-some-work')
def do_some_work():
    return service.do_some_work()

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

问题是如何手动重新加载配置文件而不重新启动Flask/gunicorn实例。

如果这个问题听起来有点奇怪，我很抱歉。我很了解经典方法：

• 通过 watchdog 或其替代方案
监控配置文件（目录）的更改
• 设置一些计时器以固定的时间间隔检查文件修改日期，例如Flask-APScheduler
• 添加额外的端点，如

  @app.route('/reload')

就我而言，它们都有各种缺点。我只是希望能够构建一些可以在不使用 HTTP 的情况下与正在运行的 Python 进程交互的 cli。但由于我不是 Python 专家，所以我不太了解 WSGI 进程隔离。

更新： 好的，答案是 RPyC，但是，gunicorn 应该仅由单个工作线程启动，或者应用程序必须使用某种缓存来存储配置数据。

Answer 1

您可以尝试使用信号处理程序。这是一个带有 SIGHUP 的示例。

# debug.py
import signal
from flask import Flask, jsonify


class Service:
    def __init__(self) -> None:
        self._config = ''
        self.load()

    def load(self):
        with open('config.txt') as f:
            self._config = f.readlines()

    def do_some_work(self):
        return self._config


app = Flask(__name__)
service = Service()


@app.route('/do-some-work')
def do_some_work():
    # just result of parsed / loaded config
    return jsonify(dict(config=service.do_some_work()))


def handler(signum, frame):
    service.load()


# add SIGHUP handler
signal.signal(signal.SIGHUP, handler)


if __name__ == '__main__':
    app.run(host='localhost', debug=True)

打开1个终端：

echo 1 >> config.txt
gunicorn -p app.pid -w 4 'debug:app'

打开2个终端：

# let's check loaded / parsed config 
curl http://localhost:8000/do-some-work
# {"config":["1\n"]}
# let's change config
echo "2345" >> config.txt
# let's send HUP
cat app.pid | xargs kill -HUP
# let's check config value
curl http://localhost:8000/do-some-work
# {"config":["1\n","2345\n"]}