最近一直在使用 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()