快捷導(dǎo)航

flask 框架操作MySQL數(shù)據(jù)庫(kù)簡(jiǎn)單示例

更新時(shí)間：2020年02月02日 11:14:31 作者：Dawn__Z

這篇文章主要介紹了flask 框架操作MySQL數(shù)據(jù)庫(kù),結(jié)合實(shí)例形式詳細(xì)分析了flask框架操作MySQL數(shù)據(jù)庫(kù)的連接、表格創(chuàng)建、數(shù)據(jù)增刪改查等相關(guān)使用技巧,需要的朋友可以參考下

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

一、創(chuàng)建數(shù)據(jù)庫(kù)表格

"""
Created on 19-10-8
@requirement:Anaconda 4.3.0 (64-bit) Python3.6
@description:創(chuàng)建表格
"""
import pymysql
server = '127.0.0.1'
user = 'root'
password = 'password'
# 連接數(shù)據(jù)庫(kù)
conn = pymysql.connect(server, user, password, database='information_collection') # 獲取連接
cursor = conn.cursor() # 獲取游標(biāo)
# "**ENGINE=InnoDB DEFAULT CHARSET=utf8**"-創(chuàng)建表的過(guò)程中增加這條，中文就不是亂碼
# 創(chuàng)建表
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
 """)
# 查詢(xún)數(shù)據(jù)庫(kù)表user內(nèi)容
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)
# # 插入數(shù)據(jù)，注：與sqlserver有些區(qū)別
cursor.execute("INSERT INTO user VALUES('%s','%s','%s','%s')" % ('xiaoming','qwe','ming','@163.com'))
# 提交數(shù)據(jù)，才會(huì)寫(xiě)入表格
conn.commit()
# 關(guān)閉游標(biāo)關(guān)閉數(shù)據(jù)庫(kù)
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 SQLAlchemy
from flask import Flask, jsonify, request
import configparser
import os
app = Flask(__name__)
# 使用ConfigParser 首選需要初始化實(shí)例，并讀取配置文件：
my_config = configparser.ConfigParser()
my_config.read('db.conf')
# 連接數(shù)據(jù)庫(kù)information_collection
app.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'] = True
mydb = SQLAlchemy()
mydb.init_app(app)
# 用戶(hù)模型
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
# 獲取用戶(hù)列表，所有數(shù)據(jù)
@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)
# 添加用戶(hù)數(shù)據(jù)，一條一條添加
@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)
# 獲取單條數(shù)據(jù)
@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': '找不到數(shù)據(jù)'}
  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)
# 修改用戶(hù)數(shù)據(jù)
@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)
# 刪除用戶(hù)數(shù)據(jù)
@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()

三、返回?cái)?shù)據(jù)的樣式