本文實例講述了flask 框架操作MySQL數據庫。分享給大家供大家參考，具體如下：

一、創建數據庫表格

"""Created on 19-10-8@requirement:Anaconda 4.3.0 (64-bit) Python3.6@description:創建表格"""import pymysqlserver = '127.0.0.1'user = 'root'password = 'password'# 連接數據庫conn = pymysql.connect(server, user, password, database='information_collection') # 獲取連接cursor = conn.cursor() # 獲取游標# "**ENGINE=InnoDB DEFAULT CHARSET=utf8**"-創建表的過程中增加這條，中文就不是亂碼# 創建表cursor.execute("""CREATE TABLE if not exists user(   user_id INT NOT NULL auto_increment primary key,   user_name VARCHAR(100),   user_password VARCHAR(100),   user_nickname VARCHAR(100),   user_email VARCHAR(100)   )   ENGINE=InnoDB DEFAULT CHARSET=utf8 """)# 查詢數據庫表user內容cursor.execute('SELECT * FROM user')# 查看一行 多行：cursor.fetchall()row = cursor.fetchone()print(row)# if row[0] is None:#   row0 = list(row)#   row0[0] = 0#   row = tuple(row0)# # 插入數據，注：與sqlserver有些區別cursor.execute("INSERT INTO user VALUES('%s','%s','%s','%s')" % ('xiaoming','qwe','ming','@163.com'))# 提交數據，才會寫入表格conn.commit()# 關閉游標關閉數據庫cursor.close()conn.close()

二、flask操作mysql

"""Created on 19-10-8@requirement:Anaconda 4.3.0 (64-bit) Python3.6@description:"""from flask_sqlalchemy import SQLAlchemyfrom flask import Flask, jsonify, requestimport configparserimport osapp = Flask(__name__)# 使用ConfigParser 首選需要初始化實例，并讀取配置文件：my_config = configparser.ConfigParser()my_config.read('db.conf')# 連接數據庫information_collectionapp.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get('DEV_DATABASE_URL') or /               "mysql+pymysql://root:password@127.0.0.1:3306/information_collection"app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = Truemydb = SQLAlchemy()mydb.init_app(app)# 用戶模型class User(mydb.Model):  user_id = mydb.Column(mydb.Integer, primary_key=True)  user_name = mydb.Column(mydb.String(60), nullable=False)  user_password = mydb.Column(mydb.String(30), nullable=False)  user_nickname = mydb.Column(mydb.String(50))  user_email = mydb.Column(mydb.String(30), nullable=False)  def __repr__(self):    return '<User %r>' % self.user_name# 獲取用戶列表，所有數據@app.route('/users', methods=['GET'])def getUsers():  data = User.query.all()  datas = []  for user in data:    datas.append({'user_id': user.user_id, 'user_name': user.user_name, 'user_nickname': user.user_nickname, 'user_email': user.user_email})  return jsonify(data=datas)# 添加用戶數據，一條一條添加@app.route('/user', methods=['POST'])def addUser():  user_name = request.form.get('user_name')  user_password = request.form.get('user_password')  user_nickname = request.form.get('user_nickname')  user_email = request.form.get('user_email')  user = User(user_name=user_name, user_password=user_password, user_nickname=user_nickname, user_email=user_email)  try:    mydb.session.add(user)    mydb.session.commit()  except:    mydb.session.rollback()    mydb.session.flush()  userId = user.user_id  if (user.user_id is None):    result = {'msg': '添加失敗'}    return jsonify(data=result)  data = User.query.filter_by(user_id=userId).first()  result = {'user_id': user.user_id, 'user_name': user.user_name, 'user_nickname': user.user_nickname, 'user_email': user.user_email}  return jsonify(data=result)# 獲取單條數據@app.route('/user/<int:userId>', methods=['GET'])def getUser(userId):  user = User.query.filter_by(user_id=userId).first()  if (user is None):    result = {'msg': '找不到數據'}  else:    result = {'user_id': user.user_id, 'user_name': user.user_name, 'user_nickname': user.user_nickname, 'user_email': user.user_email}  return jsonify(data=result)# 修改用戶數據@app.route('/user/<int:userId>', methods=['PATCH'])def updateUser(userId):  user_name = request.form.get('user_name')  user_password = request.form.get('user_password')  user_nickname = request.form.get('user_nickname')  user_email = request.form.get('user_email')  try:    user = User.query.filter_by(user_id=userId).first()    if (user is None):      result = {'msg': '找不到要修改的記錄'}      return jsonify(data=result)    else:      user.user_name = user_name      user.user_password = user_password      user.user_nickname = user_nickname      user.user_email = user_email      mydb.session.commit()  except:    mydb.session.rollback() # 回滾    mydb.session.flush() # 重置  userId = user.user_id  data = User.query.filter_by(user_id=userId).first()  result = {'user_id': user.user_id, 'user_name': user.user_name, 'user_password': user.user_password, 'user_nickname': user.user_nickname, 'user_email': user.user_email}  return jsonify(data=result)# 刪除用戶數據@app.route('/user/<int:userId>', methods=['DELETE'])def deleteUser(userId):  User.query.filter_by(user_id=userId).delete()  mydb.session.commit()  return getUsers()if __name__ == '__main__':  app.run()