当前位置：首页 > news >正文

python sqlalchemy（ORM）- 03 增删改查

news 2025/8/3 4:35:45

文章目录

ORM更新数据
ORM查询
ORM删除操作
处理关系对象
多表的关联查询

本节所有案例基于（第一节 python sqlalchemy（ORM）- 01 ORM简单使用）中的User、Address两个模型类

ORM更新数据

查询到模型类对象，直接修改其属性值，即可更新；
查看更新的实例 session.dirty ；
查看新添加的实例对象–> session.new；
新添加的对象在session 中，user_obj in session ->True
只有提交事务session.commit()，更新、添加才会同步到数据库中。
回滚事务session.rollback()，则所有的修改、添加会恢复到之前的状态；

ORM查询

查询所有的字段，获取 Query 对象

from sqlalchemy.orm.query import Query
from sqlalchemy import text# 查询User的所有字段，并按照id升序排序
query_obj = session.query(User).order_by(text("id asc"))# 获取query对象中的数据
# 1. 遍历 获取每一个（记录）对象
# 2. query_obj.first()/.all()/.one()/.one_or_none() 获取对象#

查询局部字段，返回Query对象

# 查询name, fullname字段
query_obj = session.query(User.name, User.fullname)
# 对于query_obj 可遍历、也可直接调用获取数据对象的方法
# 结果[('tom', '李四'), ('jack', '张三')]# 查询User所有字段 + name字段
query_obj = session.query(User, User.name).all()
# 元组列表 [(tom, 'tom'), (jack, 'jack')]
# 元组第一个元素为User对象，第二个为name值# 为字段指定别名  相当于select name as user_name from user_t;
query_obj = session.query(User.name.label("user_name")).all()# 为模型类 指定别名
from sqlalchemy.orm import aliased
user_alias = aliased(User, name="user_alias")# 根据别名查询
query_obj = session.query(user_alias).all()

排序、分页

结果分片，实现分页；
offset().limit() 实现分页


# 查询User的所有字段的数据，按照id降序排序，并分页获取第一条
result = session.query(User).order_by(-User.id)[:1]  # 分片 分页
# 按照id升序排序，并分页获取第二条数据
result2 = session.query(User).order_by(User.id).offset(1).limit(1).all()

过滤查询

filter 复杂过滤，条件如User.id > 3 or text(“id > 3”)
filter_by 简单的等值过滤；

# 等值过滤
session.query(User.name).filter_by(fullname="Ed Jones")
# 复杂过滤
session.query(User.name).filter(User.fullname == "Ed Jones")
# 链式过滤
session.query(User).filter(User.name == "ed").filter(User.fullname == "Ed Jones")

过滤操作符
- _eq_() 等值匹配
- _ne_() 不等
- like 模糊匹配（区分大小写） % 多个字符的通配符
- ilike 不区分大小写
- in_([]) 范围匹配
- not_in([]) 不在该范围，相当于 ~ xx.in_
- is_() 是None
- is_not() 不是None
- and_(条件1，条件2，…) 多个条件的与
- or_(条件1，条件2，…) 多个条件的或
- ~ 条件的取反
- match() 包含xxx sqlite数据库不支持

# __eq__()
user = session.query(User).filter(User.id==1).one()
# 等价于
user = session.query(User).filter(User.id.__eq__(1)).one()
# __ne__ 类似的用法# like 模糊查询（区分大小写）  % 多个字符通配符    ilike 不分大小写
user_list = session.query(User).filter(User.name.like("lau%")).all()# in_() 列表范围查询/query对象   not_in()
query.filter(User.name.in_(["ed", "wendy", "jack"])).all()
# 嵌套查询
query.filter(User.name.in_(session.query(User.name).filter(User.name.like("%ed%"))))# use tuple_() for 复合查询， 多列一起  in_()
from sqlalchemy import tuple_
query.filter(tuple_(User.name, User.nickname).in_([("ed", "edsnickname"), ("wendy", "windy")])).all()# is_()  is_not()
query.filter(User.name == None)
# 等价于
query.filter(User.name.is_(None))# use and_()  多个条件与
from sqlalchemy import and_
query.filter(and_(User.name == 'ed', User.fullname == 'Ed Jones'))
# 等价于
query.filter(User.name == 'ed', User.fullname == 'Ed Jones')
# 链式
query.filter(User.name == 'ed').filter(User.fullname == 'Ed Jones')# 多个条件或
from sqlalchemy import or_
query.filter(or_(User.name == 'ed', User.name == 'wendy'))

使用文本sql 过滤、排序等。

from sqlalchemy import text# 查询出 id < 3 的所有数据，并按照id 升序排序
session.query(User).filter(User.id<3).order_by(User.id).all()
# text sql
session.query(User).filter(text("id<3")).order_by(text("id asc")).all()# 参数化的文本sql
# 查询id > 3 且 name like lau% 的所有数据
session.query(User).filter(text("id<:id and name ilike :name")).params(
id=3, name="lau%").order_by(-User.id).all()# 完整的text sql -> from_statement()  后面不能再跟条件
session.query(User).from_statement(text("SELECT * FROM users where name=:name")).params(name="ed").all()
# text sql 指定字段
stmt = text("SELECT name, id, fullname, nickname FROM users where name=:name")
stmt = stmt.columns(User.name, User.id, User.fullname, User.nickname)
session.query(User).from_statement(stmt).params(name="ed").all()

6. 聚合查询

# 统计查询的条数 query.count()
session.query(User).filter(User.name.like("%ed")).count()# 聚合函数
from sqlalchemy import func# 分组查询，label指定别名
r = session.query(func.count(User.sex).label("num"), User.sex.label("user_sex")).group_by(User.sex).all()# 统计行数
session.query(func.count("*")).select_from(User).scalar() # scalar() 获取第一行的第一列的值

更多查询参考

ORM删除操作

pass

处理关系对象

当创建一个User对象时，它的addresses 关系属性(非表字段)是一个空列表；当然也可以配置为集合、或者字典等，方法参考。
当添加User对象（主表记录）时，addresses列表中的Address对象会自动添加（子表记录自动添加）。
操作案例

# 创建User对象
jack = User(name="jack", fullname="Jack Bean", sex="male")
jack.addresses
# []# 添加地址
jack.addresses.append(Address(email_address="北京天安门"))
jack.addresses.append(Address(email_address="山东济南"))
# 一旦添加一个地址对象，用户、地址对象就建立双向关系
address0 = jack.addresses[0]
address0.user  # 可以获取 jack 用户对象# 只需将jack对象，添加到数据库中，地址自动添加
session.add(jack)
session.commit()

当查询用户对象时，仅仅返回一个用户对象，并不处理addresses关系的查询，当用户对象调用addresses属性时，突然就触发地址对象的sql查询。

多表的关联查询

字段的等值多虑；

result = session.query(User, Address).filter(User.id == Address.user_id) # 字段等值过滤 连接.all()# [(jack, 北京), (jack, 河南), (tom, 武汉)]
# 对应sql
SELECT user_t.id AS u_id,user_t.name AS u_name,user_t.sex AS u_sex,address_t.id AS addr_id,address_t.email_address AS addr_email_address,address_t.user_id AS addr_user_id
FROM user_t, address_t
WHERE user_t.id = address_t.user_idAND address.email_address = ?

query.join() 关系连接

# 有一个外键的情况下
session.query(User) # 仅查询User中的数据.join(Address) # 自动根据外键连接.filter(Address.email_address == "北京").all()
# [jack]
session.query(User, Address) # 仅查询User中的数据.join(Address) # 自动根据外键连接.filter(Address.email_address == "北京").all()
# [(jack, 北京)]# 有多个外键或者无外键， 使用关系连接
result = session.query(User, Address) # 查询两张表，返回Query对象.join(User.addresses) # 通过User.addresses 关系 连接两张表.filter(Address.email_address.match("北京")) # query对象过滤  包含‘北京’.all() # 获取所有的数据
# [(jack, 北京)]  两个对象的元组  列表

关系连接，同时指定条件

r = session.query(User, Address).join(User.addresses.and_(~Address.title.match("北京"))) # 指定关系连接，同时指定过滤条件，地址不包含‘北京’.all()# [(jack, 河南), (tom, 武汉)]

查询的多个表中，默认从左到右依次连接，若要指定连接的第一个表，可以使用query.select_from(xxx)

query = session.query(User, Address) # 查询多个表.select_from(Address) # Address作为起始表  开始连接.join(Address.user) # 关系连接.filter(User.name.match("jack")) # 条件过滤.all() 
# [(jack, 北京), (jack, 河南)]# 左外连接
result = session.query(User, Address).outerjoin(User.addresses) # 左连接  以左边表的所有记录为主，可多 不可少.all()# [(jack, 北京), (jack, 河南), (tom, 武汉), (lauf, None), (LAUF, None), (laufing, None)]

连接指定的表

from sqlalchemy.orm import aliasedadd1 = aliased(Address, name="add1")
add2 = aliased(Address, name="add2")result = session.query(User, add1)
.join(add1, User.addresses) # 连接add1  相当于 join(User.addresses.of_type(add1))
.join(add2, User.addresses) # 连接add2
.all()