Python 脚本与将内容传输到客户端的 Python Flask 服务器通信的最佳方式是啥？

Posted 2023-02-26

【中文标题】Python 脚本与将内容传输到客户端的 Python Flask 服务器通信的最佳方式是啥？

【英文标题】：What is the best way for a Python script to communicate with a Python Flask server that transports content to the client?Python 脚本与将内容传输到客户端的 Python Flask 服务器通信的最佳方式是什么？

【发布时间】：2020-11-22 17:05:48

【问题描述】：

以下场景：

我有一个 Raspberry Pi 作为服务器运行。目前我正在使用带有 Flask 的 Python 脚本，我也可以从我的 PC 访问 Raspberry Pi。 （flask 服务器运行一个 react 应用程序。）

但功能应该被扩展。它应该如下所示：

第二个 Python 脚本一直在运行。此 Python 脚本每秒从外部 API 获取数据并进行处理。如果满足某些条件，则应处理数据，然后将数据传送到 Python Flask 服务器。然后 Flask 服务器将数据转发到计算机上运行的网站。

如何或哪种方法最适合对这种“进程间通信”进行编程。有图书馆吗？我尝试过 Celery，但每当我想访问外部 API 时，它都会抛出我的第二个 Python 脚本，所以我不知道这是否是正确的选择。

还有什么最好的方法？穿线？直接进程间通信？

如果重要的话，这就是我的服务器应用程序到目前为止的样子：

from gevent import monkey
from flask import Flask, render_template
from flask_socketio import SocketIO
monkey.patch_all()

app = Flask(__name__, template_folder='./build', static_folder='./build/static')
socket_io = SocketIO(app)


@app.route('/')
def main():
    return render_template('index.html')


@socket_io.on('fromFrontend')
def handleInput(input):
    print('Input from Frontend: ' + input)
    send_time()


@socket_io.on('time')
def send_time():
    socket_io.emit('time', 'returnTime': "some time")


if __name__ == '__main__':
    socket_io.run(app, host='0.0.0.0', port=5000, debug=True)

【参考方案1】：

好吧，我找到了针对我的特定问题的解决方案，我用一个线程实现了它，如下所示：

import gevent.monkey
gevent.monkey.patch_all()


from flask import Flask, render_template
from flask_socketio import SocketIO
import time
import requests
from threading import Thread


app = Flask(__name__, template_folder='./build', static_folder='./build/static')
socket_io = SocketIO(app)


@app.route('/')
def main():
    thread = Thread(target=backgroundTask)
    thread.daemon = True
    thread.start()
    return render_template('index.html')


@socket_io.on('fromFrontend')
def handleInput(input):
    print('Input from Frontend: ' + input)

@socket_io.on('time')
def send_time():
    socket_io.emit('time', 'returnTime': 'hi frontend')


def backgroundTask():
    # do something here
    # access socket to push some data
    socket_io.emit('time', 'returnTime': "some time")


if __name__ == '__main__':
    socket_io.run(app, host='0.0.0.0', port=5000, debug=True)