Flask-MySQLdb是一个Flask扩展,可用于在Flask应用程序中使用MySQL数据库。它提供了一些方便的方法和功能,以便在Flask应用程序中轻松地连接和操作MySQL数据库。
1、安装Flask-MySQLdb库
pip install flask-mysqldb
注意:在Windows系统中直接安装即可,亲测在Centos7系统中安装失败,所需要的依赖没有安装成功,可以使用pymysql库代替此库。
2、导入Flask-MySQLdb库
在Flask应用程序中导入Flask-MySQLdb库:
from flask_mysqldb import MySQL
3、配置MySQL数据库连接
在Flask应用程序中配置MySQL数据库连接:
app.config['MYSQL_HOST'] = 'localhost' # MySQL主机地址
app.config['MYSQL_USER'] = 'username' # MySQL用户名
app.config['MYSQL_PASSWORD'] = 'password' # MySQL密码
app.config['MYSQL_DB'] = 'database' # MySQL数据库名
4、初始化MySQL对象
在Flask应用程序中初始化MySQL对象:
mysql = MySQL(app)
5、使用MySQL对象查询数据
使用MySQL对象的cursor方法执行SQL查询并获取数据:
cur = mysql.connection.cursor()
cur.execute("SELECT * FROM users")
data = cur.fetchall()
6、使用MySQL对象插入数据
使用MySQL对象的cursor方法执行SQL插入操作:
cur = mysql.connection.cursor()
cur.execute("INSERT INTO users (name, email) VALUES (%s, %s)", ("John", "[email protected]"))
mysql.connection.commit()
7、关闭MySQL连接
使用MySQL对象的close方法关闭MySQL连接:
mysql.connection.close()
8、Flask框架应用flask-mysqldb库
示例代码:
from flask import Flask, request, jsonify
from flask_mysqldb import MySQL
app = Flask(__name__)
app.config['MYSQL_HOST'] = "39.xx.58.xx"
app.config['MYSQL_USER'] = "root"
app.config['MYSQL_PASSWORD'] = "dxxxxx"
app.config['MYSQL_DB'] = "flxxxxx"
mysql = MySQL(app)
@app.route("/users", methods=["GET"])
def get_all_users():
query = request.args.get("query", None)
pagenum = request.args.get("pagenum", 1)
pagesize = request.args.get("pagesize", 5)
if not query:
cur = mysql.connection.cursor()
cur.execute(f'SELECT * FROM users limit {(int(pagenum) - 1) * int(pagesize)}, {pagesize}')
datas = cur.fetchall()
cur.execute(
f"SELECT count(*) FROM users")
total = cur.fetchall()
if total:
total = total[0][0]
else:
total = 0
datas_list = []
for data in datas:
dic = {"id": data[0], "username": data[1], "password": data[2], "age": data[3], "create_time": data[4], "role_name": data[5], "mobile": data[6], "email": data[7], "mg_state": data[8]}
if dic.get('mg_state') == 1:
dic['mg_state'] = True
else:
dic['mg_state'] = False
datas_list.append(dic)
cur.close()
return jsonify({"meta": {'msg': 'success', 'status': 200}, "data": {"users": datas_list, "total": total, "pagenum": pagenum}})
else:
cur = mysql.connection.cursor()
cur.execute(f"SELECT * FROM users where username like '%{query}%' limit {(int(pagenum) - 1) * int(pagesize)}, {pagesize}")
datas = cur.fetchall()
cur.execute(
f"SELECT count(*) FROM users where username like '%{query}%'")
total = cur.fetchall()
if total:
total = total[0][0]
else:
total = 0
datas_list = []
for data in datas:
dic = {"id": data[0], "username": data[1], "password": data[2], "age": data[3], "create_time": data[4],
"role_name": data[5], "mobile": data[6], "email": data[7], "mg_state": data[8]}
if dic.get('mg_state') == 1:
dic['mg_state'] = True
else:
dic['mg_state'] = False
datas_list.append(dic)
cur.close()
return jsonify({"meta": {'msg': 'success', 'status': 200},
"data": {"users": datas_list, "total": total, "pagenum": pagenum}})
@app.route('/users', methods=['POST'])
def add_user():
json_data = request.get_json()
username = json_data.get('username')
password = json_data.get('password')
age = json_data.get('age')
create_time = json_data.get('create_time')
role_name = json_data.get('role_name')
mobile = json_data.get('mobile')
email = json_data.get('email')
cur = mysql.connection.cursor()
cur.execute('INSERT INTO users (username, password, age, create_time, role_name, mobile, email) '
'VALUES (%s, %s, %s, %s, %s, %s, %s)', (username, password, age, create_time, role_name, mobile, email))
mysql.connection.commit()
cur.close()
data = json_data
return jsonify({"meta": {'msg': 'success', 'status': 200}, "data": data})
@app.route('/users/<int:id>', methods=['PUT'])
def update_user(id):
json_data = request.get_json()
mobile = json_data.get('mobile')
email = json_data.get('email')
cur = mysql.connection.cursor()
cur.execute('UPDATE users SET mobile=%s, email=%s WHERE id=%s', (mobile, email, id))
mysql.connection.commit()
cur.close()
return jsonify({"meta": {'msg': 'success', 'status': 200}, "data": json_data})
@app.route('/users/<int:id>', methods=['DELETE'])
def delete_user(id):
cur = mysql.connection.cursor()
cur.execute('DELETE FROM users WHERE id=%s', (id,))
mysql.connection.commit()
cur.close()
data = ""
return jsonify({"meta": {'msg': '删除成功', 'status': 200}, "data": data})
if __name__ == '__main__':
app.run()
运行结果:
GET:
POST:
PUT:
DELETE: