最近一直在使用 Flask 进行我们的质量管理平台后端的API 开发, 数据库相关我使用了 Flask-SQLAlchemy, 下面是在开发过程中, 学习总结的一些关于 SQLAlchemy 的一些知识.

首先附上在 models 定义的关于 User 的数据库表, 定义模型.

from flask_security import RoleMixin, UserMixin
from config import db
from sqlalchemy_utils import PasswordType

# User 是用户表, Role 是用户角色表, roles_users 是两表的关联表
roles_users = db.Table('roles_users',
                       db.Column('user_id', db.Integer(), db.ForeignKey('user.id')),
                       db.Column('role_id', db.Integer(), db.ForeignKey('role.id')))

class Role(db.Model, RoleMixin):
    id = db.Column(db.Integer(), primary_key=True)
    name = db.Column(db.String(80), unique=True)
    description = db.Column(db.String(255))

class User(db.Model, UserMixin):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True)
    email = db.Column(db.String(255), unique=True)
    password = db.Column(PasswordType(
        schemes=[
            'pbkdf2_sha512',
            'md5_crypt'
        ],

        deprecated=['md5_crypt']
    ))
    active = db.Column(db.Boolean(), default=True)
    confirmed_at = db.Column(db.DateTime(), default=db.func.now())
    # 设置两表关联
    roles = db.relationship('Role', secondary=roles_users,
                            backref=db.backref('users', lazy='dynamic'), lazy='dynamic')

配置 Flask-SQLAlchemy.

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask.ext.restful import Api
from flasgger import Swagger

app = Flask(__name__)
api = Api(app)
swagger = Swagger(app)
#配置数据库地址
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:root@dbhost:3306/holmes'
app.config['SQLALCHEMY_ECHO'] = False
app.config['SQLALCHEMY_POOL_SIZE'] = 20
app.config['SQLALCHEMY_MAX_OVERFLOW'] = 50
db = SQLAlchemy(app)

最后使用 db.create_all() 来创建数据库. OK, 下面来说一下数据库相关的操作

Flask-SQLAlchemy 增删改查

先将一些需要的 import 进来.

from models.user import User, Role
from config import db
from flask import request

插入一个用户

def addUser(req_data):
    # 根据用户名, 判断用户是否已经存在
    if User.query.filter_by(username=req_data['username']) is not None:
        return {'msg': '用户已存在'}
        
    # 创建一个新用户
    user = User()
    user.username = req_data['username']
    user.email = user.username + "@4paradigm.com"
    user.password = req_data['password']
    for user_id in req_data['roles']:
        role = Role.query.filter(Role.id == user_id).first()
        user.roles.append(role)
    
    # 将新创建的用户添加到数据库会话中
    db.session.add(user)
    # 将数据库会话中的变动提交到数据库中, 记住, 如果不 commit, 数据库中是没有变化的.
    db.session.commit()

查找用户

# 查询所有用户
users_list = User.query.all()

# 查询用户名称为 test 的第一个用户, 并返回用户实例, 因为之前定义数据库的时候定义用户名称唯一, 所以数据库中用户名称为 test 的应该只有一个.
user = User.query.filter_by(username = 'test').first()
# or
user = User.query.filter(User.username == 'test').first()
# 查询用户 id, 用户 email, 用户 roles 类型

# 模糊查询, 查找用户名以abc 结尾的所有用户
users_list = User.query.filter(User.username.endsWith('c')).all()

#查询用户名不是 test 的第一个用户
user = User.query.filter(User.username != 'test').first()

修改一个用户的信息

# 获取用户对象
user = User.qurey.filter_by(id = user_id).first()

# 修改用户信息
user.username = 'test_update'

# 提交数据库会话
db.session.commit()

删除一个用户的信息

# 获取用户对象
user = User.qurey.filter_by(id = user_id).first()

# 删除用户
db.session.delete(user)

# 提交数据库会话
db.session.commit()