彩笔激流勇退。

1. 简介

ORM，对象关系映射。简单来说，ORM将数据库中的表与面向对象中的类建立了一种对应关系。这样，我们要操作数据库，表，记录就可以直接通过操作类或者类实例来完成。

SQLAlchemy 是目前python中最垃圾的 ORM框架, 功能全面, 使用复杂。

Flask-SQLAlchemy 是一个为 Flask 应用增加 SQLAlchemy 支持的扩展，把原本pymysql几句话就能搞定的事情整成抽象的，继承的，封装的，多态的更适合高级程序员体质的负离子保温杯。

抛开兼容性不谈，Flask-SQLAlchemy无疑是磨练程序员改bug的磨刀石，是检验程序员记忆能力的试金石，是凝结了人类精华的草酸钙结石。

常用字段类型

类型名	python接收类型	mysql生成类型	说明
Integer	int	int	整型
Float	float	float	浮点型
Numeric(5,2)	decimal.Decimal	decimal(5,2)
Boolean	bool	tinyint	整型，只占1个字节
Text	str	text	文本类型，最大64KB
LongText	str	longtext	文本类型，最大4GB
String	str	varchar	变长字符串，必须限定长度
Date	datetime.date	date	日期
DateTime	datetime.datetime	datetime	日期和时间
Time	datetime.time	time	时间
TIMESTAMP	datetime.datetime	TIMESTAMP	时间戳，可以用text(‘now()’)赋值

常用的字段选项

选项名	说明
primary_key	True，则该字段为表的主键，默认自增
unique	True，则这列设置唯一
nullable	False，则这列设置非空
default	为这列设置默认值，不作用在数据库
server_default	值必须是字符串格式，作用在数据库
index	True，则为这列创建索引，提高查询效率

如果没有给对应字段的类属性设置default参数, 且添加数据时也没有给该字段赋值, 则sqlalchemy会给该字段设置默认值 None。

常见命令

db.create_all() #创建所有表
db.drop_all() #删除所有表

2. 创建表

pip install pymysql

pip install flask-sqlalchemy

数据库URL(连接地址)格式: 协议名://用户名:密码@数据库IP:端口号/数据库名

main.py

​ 在下面代码中，我们使用了 with app.app_context(): 语句来确保当前应用实例的操作db.create_all() 是在flask应用上下文中被调用的。

from app import *
from models import User@app.route('/',methods=['GET','POST'])
def login():print(db)return 'hello world'if __name__ == '__main__':with app.app_context():db.create_all()#创建表app.run(host='0.0.0.0',port=9901,debug=1)

modules.py

表名默认为类名小写, 可以通过 __tablename__类属性 进行修改

from app import dbclass User(db.Model):# User表__tablename__ = 't_user'id= db.Column(db.Integer,primary_key=True) # 必须要有主键存在name=db.Column(db.String(20),nullable=True)# 可空age=db.Column(db.SmallInteger)gender=db.Column(db.Boolean)birthday=db.Column(db.Date)

对应MySQL语句

CREATE TABLE `t_user` (`id` int NOT NULL AUTO_INCREMENT,`name` varchar(20) COLLATE utf8mb4_general_ci DEFAULT NULL,`age` smallint DEFAULT NULL,`gender` tinyint(1) DEFAULT NULL,`birthday` date DEFAULT NULL,PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;

app.py

from flask import Flask,url_for,request,render_template,make_response,redirect,jsonify
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__) # 用本脚本名实例化Flask对象
# 设置数据库连接地址
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:123456@127.0.0.1:3306/test1'# 是否追踪数据库修改(开启后会触发一些钩子函数)  一般不开启, 会影响性能
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False# 是否显示底层执行的SQL语句
app.config['SQLALCHEMY_ECHO'] = True# 初始化组件对象, 直接关联Flask应用
db = SQLAlchemy(app)

3. 数据表简单查询

user=User.query

	说明
user.filter_by(id=1)	只能等值查询,使用=
user.filter(User.id==1)	条件查询，用==

use.filter 参数与运算符	说明
and_(User.id==1,User.age==99)	与
or_(User.id==1,User.age==99)	或
~(User.id==1)	非
!= None，>，>=，
User.name.like('%a%')	模糊查询
User.id.in_((1,2,5))	范围查询
User.id.between(1,3)	[1,3]

if __name__ == '__main__':with app.app_context():db.create_all()user=User.queryprint(user)#显示sql语句, 返回Query对象print(user.all())#[<User 1>, <User 2>, <User 3>],每个元素都是models.User类型print(user.count()) #返回query中的Model对象数量print(user.filter_by(age=30,id=1)) #显示SQL语句, 返回Query对象，内部条件为交集

4. 映射查询 db.session.query

映射查询在SQLAlchemy中，可以通过session对象的query方法完成。

注意关键字书写顺序

db.session.query().filter().group_by().having().order_by().paginate().all()

if __name__ == '__main__':with app.app_context():db.create_all()query=db.session.query(User) #<class 'flask_sqlalchemy.query.Query'>print(query.filter(User.id>1).all())# [<User 2>, <User 3>]query2=db.session.query(User.id,User.name)print(query2.filter(User.age==99).all())# [(2, 'tom')]app.run(host='0.0.0.0',port=9901,debug=1)

5. 排序 order_by

from sqlalchemy import desc

if __name__ == '__main__':with app.app_context():db.create_all()query = db.session.query(User.id, User.name,User.age)print(query.order_by(User.age).all())  # 默认升序排序,asc()print(query.order_by(desc(User.age)).all())  # 降序排序print(query.order_by(User.age,User.id).all())  # 先排age，后排idapp.run(host='0.0.0.0', port=9901, debug=1)# 看起来是一个阻塞函数

6. 聚合函数

数据库先添加一个age为30的记录。

from sqlalchemy import func

聚合函数	说明
count()	记录数量
sum()	加和总值
avg()	平均值
max()	最大值
min()	最小值

if __name__ == '__main__':with app.app_context():db.create_all()query = db.session.query(func.max(User.age),func.avg(User.age))#相当于 SELECT max(age) , avg(age) FROM t_userprint(query.all())  # [(99, Decimal('51.0000'))]app.run(host='0.0.0.0', port=9901, debug=1)

7. 分组查询 group_by

if __name__ == '__main__':with app.app_context():db.create_all()query = db.session.query(func.count(User.id))print(query.group_by(User.age).all())  # [(2,), (1,), (1,)]app.run(host='0.0.0.0', port=9901, debug=1)

8. 增删改

flask开了debug模式，删除数据会导致main函数重新执行，给爷整笑了。

if __name__ == '__main__':with app.app_context():db.create_all()#更新u = db.session.query(User.id==1)# 查询主键为1的记录u.name="Jack"#db.session.rollback() 事务回滚，默认遇到错误自动回滚db.session.commit()# 事务提交#删除u2=db.session.query(User).filter(User.id==6).all() # User 模型的实例if len(u2)!=0:db.session.delete(u2[0])db.session.commit()  # 事务提交#增加u3=User(id=7,name="lihua",age=35,gender=1,birthday='2077-1-1')db.session.add(u3)#db.session.add_all([u1,u2,u3]) 一次添加多个db.session.commit()app.run(host='0.0.0.0', port=9901, debug=0)# debug=1时，上面的delete操作会执行多次# 大概是是检测到了文件变化重启了一次main函数???

9. 分页查询 paginate

分页查询不老老实实用limit，非要整个paginate装什么高大上。

if __name__ == '__main__':with app.app_context():db.create_all()pg=db.session.query(User).paginate(page=2,per_page=2)# QueryPagination objectprint(pg.items)#[<User 3>, <User 4>],当前页数据print(pg.pages) #3 ,一共三页for i in pg.iter_pages(): #迭代Pagination.iter_pages对象print(i)#1 2 3app.run(host='0.0.0.0', port=9901, debug=0)

10. 原生sql支持

if __name__ == '__main__':with app.app_context():db.create_all()statement=text('select * from t_user where id> :id').params(id=1)query=db.session.query(User).from_statement(statement)print(query.all())# [<User 2>, <User 3>, <User 4>, <User 7>]#最傻逼的地方来了，新版本下面语句不支持# statement2 = text('select max(id) as mmid,max(age) as mage from t_user where id> :id').params(id=2)# query2 = db.session.query('mmid','mage').from_statement(statement2)# sqlalchemy.exc.ArgumentError# 感觉不如直接pymysqlsql=text('select max(id) as mmid,max(age) as mage from t_user where id> :id')result=db.session.execute(sql,{'id':2})# CursorResult object#print(result.fetchall())# [(7, 45)] 如果这里获取了，下面就获取不了，有点类似游标后移导致没数据读for i in result:print(i)# (7, 45)app.run(host='0.0.0.0', port=9901, debug=0)

参考

flask框架与mysql开发入门到实践 白菜爱科技