彩笔激流勇退。
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开发入门到实践 白菜爱科技
