- 存在外键
- 指定字段会返回
sqlalchemy.engine.row.Row
- 不指定字段返回
Model
,相当于全字段查询,浪费I/O - 指定字段查询建议加在 query中,不建议使用
.with_entities
,以后版本可能删除
- 不指定字段返回
- .first() 会在SQL结尾添加
limit 1
,确保返回一行- 不推荐使用
.one()
,返回超过一行会抛出异常
- 不推荐使用
# 1.不指定字段查询,不推荐
model_obj = db.session.query(Xxx).select_from(Xxx). \
filter(Xxx.xxx_id == 'a872e3f879d13e799acf7ef724eaee21').first()
"""实际执行SQL
SELECT t_xxx.id AS t_xxx_id
,t_xxx.xxx_id AS t_xxx_xxx_id
,t_xxx.xxx_type AS t_xxx_xxx_type
,t_xxx.xxx_name AS t_xxx_xxx_name
,t_xxx.xxx_name_bin AS t_xxx_xxx_name_bin
,t_xxx.xxx_price AS t_xxx_xxx_price
,t_xxx.is_deleted AS t_xxx_is_deleted
,t_xxx.create_time AS t_xxx_create_time
,t_xxx.update_time AS t_xxx_update_time
FROM t_xxx
WHERE t_xxx.xxx_id = 'a872e3f879d13e799acf7ef724eaee21'
LIMIT 1
"""
# 2.指定字段查询,推荐
row = db.session.query(Xxx.xxx_id, Xxx.xxx_name).select_from(Xxx). \
filter(Xxx.xxx_id == 'a872e3f879d13e799acf7ef724eaee21').first()
"""实际执行SQL
SELECT t_xxx.xxx_id AS t_xxx_xxx_id
,t_xxx.xxx_name AS t_xxx_xxx_name
FROM t_xxx
WHERE t_xxx.xxx_id = 'a872e3f879d13e799acf7ef724eaee21'
LIMIT 1
"""
- 返回类型:
List[sqlalchemy.engine.row.Row]
result = db.session.query(Xxx.xxx_id, Xxx.xxx_name).select_from(Xxx).all()
"""实际执行SQL
SELECT t_xxx.xxx_id AS t_xxx_xxx_id
,t_xxx.xxx_name AS t_xxx_xxx_name
FROM t_xxx
"""
- .offset(m).limit(n) 对应mysql的
limit m, n
- SQLAlchemy切片与django ORM切片实现原理不同
- django ORM切片是真分页
- SQLAlchemy的切片是假分页,逻辑层数据切片
result = db.session.query(Xxx.xxx_id, Xxx.xxx_name).select_from(Xxx).offset(11).limit(2).all()
"""实际执行SQL
SELECT t_xxx.xxx_id AS t_xxx_xxx_id
,t_xxx.xxx_name AS t_xxx_xxx_name
FROM t_xxx
LIMIT 11, 2
"""
- 对应mysql的
select 1 ... limit 1
- 仅返回一个自定义标量
1
,I/O消耗极低
if db.session.query(db.literal(1)).select_from(Xxx).filter(Xxx.id >= 100).first():
pass
"""实际执行SQL
SELECT 1 AS anon_1
FROM t_xxx
WHERE t_xxx.id >= 100
LIMIT 1
"""
# 正确写法
db.session.query(Xxx.xxx_name).select_from(Xxx).filter(Xxx.xxx_name != None).limit(10).all()
"""实际执行SQL
SELECT t_xxx.xxx_name AS t_xxx_xxx_name
FROM t_xxx
WHERE t_xxx.xxx_name IS NOT NULL
LIMIT 10
"""
# 错误写法
db.session.query(Xxx.xxx_name).select_from(Xxx).filter(Xxx.xxx_name is not None).limit(10).all()
"""实际执行SQL
SELECT t_xxx.xxx_name AS t_xxx_xxx_name
FROM t_xxx
WHERE true = 1
LIMIT 10
"""
- INNER JOIN:
.join()
# 定义表别名
x1 = db.aliased(Xxx, name='x1')
xd1 = db.aliased(XxxDetail, name='xd1')
# 1.不推荐,利用sqlalchemy外键自动关联
db.session.query(x1.xxx_id, xd1.name_list).select_from(x1). \
join(xd1).filter(x1.id >= 1).limit(10).all()
# 2.推荐,手动指定关联关系
db.session.query(x1.xxx_id, xd1.name_list).select_from(x1). \
join(xd1, x1.xxx_id == xd1.xxx_id).filter(x1.id >= 1).limit(10).all()
"""实际执行SQL
SELECT x1.xxx_id AS x1_xxx_id
,xd1.name_list AS xd1_name_list
FROM t_xxx AS x1
INNER JOIN t_xxx_detail AS xd1
ON x1.xxx_id = xd1.xxx_id
WHERE x1.id >= 1
LIMIT 10
"""
- LEFT JOIN:
.outerjoin()
x1 = db.aliased(Xxx, name='x1')
xd1 = db.aliased(XxxDetail, name='xd1')
# 使用 db.and_()
db.session.query(x1.xxx_id, x1.xxx_name, xd1.name_list).select_from(x1). \
outerjoin(xd1, db.and_(x1.xxx_id == xd1.xxx_id, x1.xxx_name != None)). \
filter(x1.id >= 1).limit(10).all()
"""实际执行SQL
SELECT x1.xxx_id AS x1_xxx_id
,x1.xxx_name AS x1_xxx_name
,xd1.name_list AS xd1_name_list
FROM t_xxx AS x1
LEFT OUTER JOIN t_xxx_detail AS xd1
ON x1.xxx_id = xd1.xxx_id
AND x1.xxx_name IS NOT NULL
WHERE x1.id >= 1
LIMIT 10
"""
- IN 子查询:
column.in_()
# sbq 为 flask_sqlalchemy.BaseQuery 对象
sbq = db.session.query(Xxx.xxx_id).select_from(Xxx).filter(Xxx.xxx_name != None)
result = db.session.query(XxxDetail.name_list).filter(XxxDetail.xxx_id.in_(sbq)).limit(10).all()
"""实际执行SQL
SELECT t_xxx_detail.name_list AS t_xxx_detail_name_list
FROM t_xxx_detail
WHERE t_xxx_detail.xxx_id IN (
SELECT t_xxx.xxx_id
FROM t_xxx
WHERE t_xxx.xxx_name IS NOT NULL
)
LIMIT 10
"""
- 标量子查询
- 语法
- [sqlalchemy>=1.4]:
query.limit(1).scalar_subquery()
- [sqlalchemy<1.4]:
query.limit(1).as_scalar()
- [sqlalchemy>=1.4]:
- 临时定义的字段建议以
udc_
为前缀- 语法:
column.label('udc_column')
- udc:
User Define Column
- 语法:
- 语法
x1 = db.aliased(Xxx, name='x1')
y1 = db.aliased(Yyy, name='y1')
# 定义标量
col_yyy_id = db.session.query(y1.yyy_id).select_from(y1).filter(y1.is_deleted == 0). \
order_by(y1.create_time.desc()).limit(1).scalar_subquery()
# 执行并获取数据
result = db.session.query(x1.xxx_id, col_yyy_id.label('udc_yyy_id')).select_from(x1). \
filter(x1.is_deleted == 0).limit(10).all()
"""实际执行SQL
SELECT x1.xxx_id AS x1_xxx_id
,(
SELECT y1.yyy_id
FROM t_yyy AS y1
WHERE y1.is_deleted = 0
AND y1.xxx_id = x1.xxx_id
ORDER BY y1.create_time DESC
LIMIT 1
) AS udc_yyy_id
FROM t_xxx AS x1
WHERE x1.is_deleted = 0
LIMIT 10
"""
-
语法
Query.exists()
- 不需要添加
limit(1)
- query中不需要指定查询的字段,都会被转化为
select 1
- 不需要添加
-
用于where判断
x1 = db.aliased(Xxx, name='x1')
y1 = db.aliased(Yyy, name='y1')
# exists 可以不加limit 1
# 如果是 not exists 使用 `~exists_sbq`
exists_sbq = db.session.query(y1).select_from(y1). \
filter(y1.is_deleted == 0, y1.xxx_id == x1.xxx_id).exists()
result = db.session.query(x1.xxx_id).filter(x1.is_deleted == 0, exists_sbq).limit(10).all()
"""实际执行SQL
SELECT x1.xxx_id AS x1_xxx_id
FROM t_xxx AS x1
WHERE x1.is_deleted = 0
AND (
EXISTS (
SELECT 1
FROM t_yyy AS y1
WHERE y1.is_deleted = 0
AND y1.xxx_id = x1.xxx_id
LIMIT 1
)
)
LIMIT 10
"""
- 用做标量子查询
- 注意:添加字段别名
udc_
- 注意:添加字段别名
exists_sbq = db.session.query(y1).select_from(y1). \
filter(y1.is_deleted == 0, y1.xxx_id == x1.xxx_id).exists()
result = db.session.query(x1.xxx_id, exists_sbq.label('udc_exists_yyy')). \
filter(x1.is_deleted == 0).limit(10).all()
"""实际执行SQL
SELECT x1.xxx_id AS x1_xxx_id
,EXISTS (
SELECT 1
FROM t_yyy AS y1
WHERE y1.is_deleted = 0
AND y1.xxx_id = x1.xxx_id
) AS udc_exists_yyy
FROM t_xxx AS x1
WHERE x1.is_deleted = 0
LIMIT 10
"""
# 获取Xxx表10条数据,优先展示未删除的,不足的由已删除的补足
x1 = db.aliased(Xxx, name='x1')
x2 = db.aliased(Xxx, name='x2')
query1 = db.session.query(x1.xxx_id, x1.xxx_name).select_from(x1).filter(x1.is_deleted == 0).limit(10)
query2 = db.session.query(x2.xxx_id, x2.xxx_name).select_from(x2).filter(x1.is_deleted == 1).limit(10)
query = query1.union_all(query2)
result = query.all()[:10]
"""实际执行SQL
SELECT anon_1.x1_xxx_id AS anon_1_x1_xxx_id
,anon_1.x1_xxx_name AS anon_1_x1_xxx_name
FROM (
(
SELECT x1.xxx_id AS x1_xxx_id
,x1.xxx_name AS x1_xxx_name
FROM t_xxx AS x1
WHERE x1.is_deleted = 0
LIMIT 10
)
UNION ALL
(
SELECT x2.xxx_id AS x2_xxx_id
,x2.xxx_name AS x2_xxx_name
FROM t_xxx AS x2
,t_xxx AS x1
WHERE x1.is_deleted = 1
LIMIT 10
)
) AS anon_1
"""