1.引包
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
2.配置连接,替换为自己的MySQL 数据库的实际用户名、密码和数据库名
app = Flask(__name__) #创建应用实列 app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:pwd_Admin@localhost:3306/db1' db = SQLAlchemy(app) #初始化Flask-SQLAlchemy 扩展
* `mysql+pymysql`:指定了使用 `pymysql` 作为连接 MySQL 的驱动。 * `root`:数据库的用户名。 * `pwd_Admin`:数据库的密码。 * `localhost`:数据库服务器的地址,这里是本地机器。 * `3306`:数据库服务器的端口,MySQL 默认端口是 3306。 * `db1`:要连接的数据库名。
3.创建模型与数据库映射
class MotorControl(db.Model): id = db.Column(db.Integer, primary_key=True, autoincrement=True) start = db.Column(db.Float) stop = db.Column(db.Float) forward = db.Column(db.Float) backward = db.Column(db.Float) fault = db.Column(db.Float) ack = db.Column(db.Float)
4.创建数据库与模型对应 (默认将bool值True存储为1,False存储为0)
5.编写路由,获取post数据
首先创建了一个新的 MotorControl
实例(这是一个 SQLAlchemy 模型,代表数据库中的一个记录),并使用从 JSON 数据中提取的值初始化它。然后,将这个新记录添加到 SQLAlchemy 的会话中,并提交会话,将新记录保存到数据库中
6.运行代码
from flask import Flask, render_template, request, redirect, jsonify
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:pwd_Admin@localhost:3306/db1'
db = SQLAlchemy(app)
class MotorControl(db.Model):
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
start = db.Column(db.Float)
stop = db.Column(db.Float)
forward = db.Column(db.Float)
backward = db.Column(db.Float)
fault = db.Column(db.Float)
ack = db.Column(db.Float)
@app.route('/store', methods=['POST'])
def store_data():
json_data = request.get_json()
start = json_data.get('0.0')
stop = json_data.get('0.1')
forward = json_data.get('0.2')
backward = json_data.get('0.3')
fault = json_data.get('0.4')
ack = json_data.get('0.5')
print(start, stop, forward, backward)
new_record = MotorControl(start=start, stop=stop, forward=forward, backward=backward, fault=fault, ack=ack)
db.session.add(new_record)
db.session.commit()
return jsonify({'message': 'Data stored successfully'}), 201
if __name__ == '__main__':
app.run(host='0.0.0.0')
7.编写发送post请求的代码,对应url到/store
import requests
start = True
stop = True
forward = True
backward = True
fault = False
ack = False
json_data = {
'0.0': start,
'0.1': stop,
'0.2': forward,
'0.3': backward,
'0.4': fault,
'0.5': ack
}
r2 = requests.post('http://127.0.0.1:5000/store', json=json_data)
print(r2.text)