Flask-SQLAlchemy CRUD

当我们习惯 Flask-SQLAlchemy 对数据库进行 CRUD 操作后，就很难再回到基于原生 SQL 的代码编写了。

我们先来看 Read 操作。用实际的例子来说明用法，数据还是之前经常用到的 emp_master ，
基于上一篇 Flask 工程的文件结构，在 models.py 中定义的 Class 如下：

class EmpMaster(db.Model):
    emp_id = db.Column(db.Integer, primary_key=True)
    gender = db.Column(db.String(10), nullable=False)
    age = db.Column(db.Integer)
    email = db.Column(db.String(50))
    phone_nr = db.Column(db.String(20))
    education = db.Column(db.String(20))
    marital_stat = db.Column(db.String(20))
    nr_of_children = db.Column(db.Integer)

    def __repr__(self):
        return '<emp_master %r>' % 'id:{}'.format(self.emp_id)

查询数据

查询所有数据

from app.models import db, EmpMaster

print (EmpMaster.query.all())

EmpMaster.query 得到的是 BaseQuery 对象的实例，对于 BaseQuery 类，SQLAlchemy 提供了一些查询方法用于获得你想要的查询对象，比如 all() 方法以 list (列表) 的形式返回所有的查询结果。

根据主键查询数据

get() 或 get_or_404() 方法根据主键来查询数据。get() 方法返回 model 的实例，如果找不到，则返回 None；而 get_or_404() 则返回 Not found 的错误响应。

EmpMaster.query.get(1001)       # 1001 是主键

筛选/过滤数据

筛选过滤数据通过 filter_by() 或 filter() 方法。假设我们想查询所有女性雇员，可以这样做：

emps = EmpMaster.query.filter_by(gender='Female').all()
if len(emps) > 0:
    for emp in emps:
        print(emp.emp_id, emp.gender, emp.email, emp.education)
else:
    print ('No data.')

filter_by() 方法返回的是 BaseQuery 对象实例，参数是关键字表达式 (keyword expression)，可以多个字段，多个字段之间为 and 关系。比如要查询所有女性，并且学历为 Bachelor 的员工：

emps = EmpMaster.query.filter_by(gender='Female', education='Bachelor11').all()
if len(emps) > 0:
    for emp in emps:
        print(emp.emp_id, emp.gender, emp.email, emp.education)
else:
    print ('No data.')

更为灵活的方式是通过 filter() 方法，与 filter_by() 方法不同的是，filter() 方法的参数是 SQL Expression。举几个例子来说明。

查找所有 education 为 Bachelor 的员工：

emps = EmpMaster.query.filter(EmpMaster.education == 'Bachelor').all()
for emp in emps:
    print (emp)

如果用 filter_by() 方法，是这样的：

emps = EmpMaster.query.filter_by(education='Bachelor').all()

看出区别来了吧。下表列示了常见条件的写法。

条件	示例
等于	query.filter(EmpMaster.education == 'Bachelor')
不等于	query.filter(EmpMaster.education != 'Bachelor')
LIKE	query.filter(EmpMaster.email.like('s%'))
IN	query.filter(EmpMaster.education.in_(['Master', 'Bachelor']))
NOT IN	query.filter(~EmpMaster.education.in_(['Master', 'Bachelor']))
IS NULL	query.filter(EmpMaster.education == None)
IS NOT NULL	query.filter(EmpMaster.education != None)

AND 条件：

EmpMaster.query.filter(EmpMaster.gender=='Female', EmpMaster.education=='Bachelor')

# 或者
from sqlalchemy import and_
emps = EmpMaster.query.filter(
    and_(EmpMaster.gender=='Female', EmpMaster.education=='Bachelor')).all()

OR 条件

from sqlalchemy import or_
emps = EmpMaster.query.filter(
  or_(EmpMaster.marital_stat=='Single', EmpMaster.nr_of_children==0))

Create

SQLAlchemy 有一个 session 对象，代表临时存储区，数据的变动提交到 session 后，需要调用 commit() 将修改提交到数据库，或者 rollback() 撤销未提交的修改。

from app.models import db, EmpMaster
from app import create_app

app = create_app()

new_emp = EmpMaster(
    emp_id = 9001,
    gender = 'Female',
    age = 18,
    email = 'test@qq.com',
    phone_nr = '13800138000',
    education = 'Master',
    marital_stat = 'Single',
    nr_of_children = 0
)

db.session.add(new_emp)
db.session.commit()

可以使用 add_all() 方法批量创建记录：

emp1 = EmpMaster(...)
emp2 = EmpMaster(...)

db.session.add_all([emp1, emp2])
db.session.commit()

Update

Update 记录需要先定位到某记录，然后对需要的字段赋值后，然后调用 commit() 方法提交修改：

emp = EmpMaster.query.get(9001)
emp.email = 'e9001@qq.com'
db.session.commit()

Delete

Delete 记录先定位到某记录，然后调用 delete() 方法发送删除指到 session，最后调用 commit() 将请求提交到数据库。

emp = EmpMaster.query.get(9001)
db.session.delete(emp)
db.session.commit()