flask 框架操作MySQL数据库简单示例

 """ Created on 19-10-8 @requirement:Anaconda 4.3.0 (64-bit) Python3.6 @description:创建表格 """ import pymysql server = '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() 

 """ 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 首选需要初始化实例，并读取配置文件： my_config = configparser.ConfigParser() my_config.read('db.conf') # 连接数据库information_collection app.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get('DEV_DATABASE_URL') or \ "mysql+pymysql://root:[email protected]:3306/information_collection" app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True mydb = 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 '' % self.user_name # 获取用户列表，所有数据 @app.route('/users', methods=['GET']) def getUsers(): data = User.query.all() datas = [] for u<strong style="color:transparent">来源gao@daima#com搞(%代@#码@网</strong>ser 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/', 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/', 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/', methods=['DELETE']) def deleteUser(userId): User.query.filter_by(user_id=userId).delete() mydb.session.commit() return getUsers() if __name__ == '__main__': app.run()