ORM增删改查

在路由中使用ORM

我们需要依靠依赖注入的方式在路由中使用ORM，在此之前需要先基于异步引擎创建异步数据库会话工厂(sqlalchemy.ext.asyncio.async_sessionmaker)并创定义依赖函数

async_sessionmaker的参数:

参数 说明 推荐配置
bind 绑定异步引擎 async_engine，必填 bind=async_engine
class_ 指定会话类，固定传 AsyncSession class_=AsyncSession
autoflush 查询前是否自动刷新缓存 False（异步项目推荐）
autocommit 是否开启自动提交事务 False（统一手动事务管理）
expire_on_commit commit 后 ORM 实例是否失效，设为 False 可减少重复查库 False
info 给会话附加自定义元数据，按需传字典 {}（按需自定义） 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 基于引擎创建会话工厂
AsyncSessionLocal = async_sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=async_engine   #引擎
    )

# 获取数据库会话依赖
async def get_db():

    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()   #异常回滚

@app.get("/check")
async def check(db = Depends(get_db)):
    if db:
        return 'Successfully!'

查询

简单查询

查询表中的所有数据

1
2
3
4
5
@app.get("/users")
async def get_user_list(db : AsyncSession = Depends(get_db)):
    orm_obj = await db.execute(select(User))   # select()内填表名，即直接继承Base的类名
    users = orm_obj.scalars().all()     #.all()返回表中的全部数据,.first()返回表中的第一个数据
    return users

根据主键查询指定数据

1
2
3
4
@app.get("/users")
async def get_user_list(db : AsyncSession = Depends(get_db)):
    res = await db.get(User,1)    # get(表名,主键值)
    return res

条件查询

sqlalchemy支持在数据库中条件查询，条件语句写在select().where()中

比较

1
2
3
4
5
@app.get('/users/{username}')
async def get_user(username: str, db: AsyncSession = Depends(get_db)):
    res = await db.execute(select(User_db).where(User_db.username == username))
    user = res.scalars().all()
    return user

除了 == 外，查询条件还包括> < >= <= !=

数据库中的字符串之间、日期之间都可用这些比较运算符比较

模糊查询

只有字符串类型支持模糊查询

模糊查询的条件写在字段.like()中，用%匹配任意多个字符，_匹配单个字符

曹%匹配所有以曹开头的字符串，如曹植、曹雪芹

曹_则匹配所有以曹开头且长度为2的字符串，曹操能匹配，曹雪芹不能匹配

1
2
3
4
5
6
#查询username属性以路径参数{username}开头的数据
@app.get('/user/{username}')
async def get_user(username:str , db: AsyncSession = Depends(get_db)):
    res = await db.execute(select(User_db).where(User_db.username.like(f"{username}%")))
    user = res.scalars().all()
    return user

或 与 非

条件查询中的或 与 非逻辑关系分别用符号| & ~表示

直接看代码：

1
2
3
4
5
@app.get("/user")
async def get_user(db: AsyncSession = Depends(get_db)):
    res = await db.execute(select(User_db).where((User_db.username == 'abcd%') | (User_db.username == 'ABCD%')))
    user = res.scalars().all()
    return user

in_()包含

与Python原生的in运算符相似：

1
2
3
4
5
@app.get("/user")
async def get_user(db: AsyncSession = Depends(get_db)):
    res = await db.execute(select(User_db).where((User_db.username.in_(['dgdz','dougunduzi']))))
    user = res.scalars().all()
    return user

func方法聚合查询

func方法可直接获取字段下所有数据的最大/最小/总和/平均值等(from sqlalchemy import func)

1
2
3
4
5
@app.get("/maxid")
async def get_max_id(db : AsyncSession = Depends(get_db)):
    res = await db.execute(select(func.max(User_db.id)))
    max_id = res.scalar()
    return max_id
方法 含义
func.max(User_db.id) 最大值
func.min(User_db.id) 最小值
func.avg(User_db.id) 平均值
func.sum(User_db.id) 求总和
func.count(User_db.id) 计数

分页查询

若数据表很长，而我们只需要查询其中的部分数据，分页查询的方式将会很好用

分页查询的写法是select().offset().limit()

简单来说就是跳过前offset条数据，取接下来的limit条：

1
2
3
4
5
6
@app.get("/get_users_of_one_page")
async def get_one_page(db : AsyncSession = Depends(get_db)):
    stmt = select(User_db).offset(5).limit(3)
    res = await db.execute(stmt)
    users = res.scalars().all()
    return users

还可以写成更直接的切片形式，以下代码是等效的：

1
2
3
4
5
6
@app.get("/get_users_of_one_page")
async def get_one_page(db : AsyncSession = Depends(get_db)):
    stmt = select(User_db).slice(5,8)
    res = await db.execute(stmt)
    users = res.scalars().all()
    return users

新增

新增单条数据

新增数据需要先实例化 ORM 模型对象，然后通过 session.add() 将对象添加到会话，最后 commit 提交事务。

add() 的参数：

参数 说明
instance ORM 模型实例，必填 
1
2
3
4
5
6
7
8
9
10
@app.post("/users")
async def create_user(db: AsyncSession = Depends(get_db)):
    new_user = User_db(                          # 实例化 ORM 模型
        username="豆棍度子",
        email="dgdz@example.com"
    )
    db.add(new_user)                             # 添加到会话
    await db.commit()                            # 提交事务
    await db.refresh(new_user)                   # 刷新得到数据库中的新数据（含自增 id）
    return new_user

refresh() 用于从数据库中重新加载实例的最新状态，通常在 commit 后调用以获取数据库自动生成的字段（如自增主键、默认值等）

通过 Pydantic 模型接收请求体

实际开发中不会把数据硬编码在代码里，而是通过请求体传入。需要先定义一个 Pydantic 模型来校验和接收数据：

1
2
3
4
5
6
from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    username: str
    email: str
    password: str

然后在路由函数中将其作为参数接收：

1
2
3
4
5
6
7
8
9
10
11
@app.post("/users")
async def create_user(user_data: UserCreate, db: AsyncSession = Depends(get_db)):
    new_user = User_db(
        username=user_data.username,
        email=user_data.email,
        password=user_data.password
    )
    db.add(new_user)
    await db.commit()
    await db.refresh(new_user)
    return new_user

新增多条数据

批量添加数据使用 session.add_all()，参数是一个 ORM 实例的可迭代对象：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@app.post("/users/batch")
async def create_users(db: AsyncSession = Depends(get_db)):
    users = [
        User_db(username="曹操", email="cc@example.com"),
        User_db(username="曹植", email="cz@example.com"),
        User_db(username="曹雪芹", email="cxq@example.com"),
    ]
    db.add_all(users)
    await db.commit()

    # 刷新所有实例以获取自增 id
    for user in users:
        await db.refresh(user)

    return users

也可以配合 Pydantic 模型接收列表：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from pydantic import BaseModel

class UserCreate(BaseModel):
    username: str
    email: str

@app.post("/users/batch")
async def create_users(users_data: list[UserCreate], db: AsyncSession = Depends(get_db)):
    users = [User_db(**u.model_dump()) for u in users_data]   # 解包字典快速创建
    db.add_all(users)
    await db.commit()

    for user in users:
        await db.refresh(user)

    return users

更新

先查后改

最常规的更新方式：先查出目标数据，再修改属性，最后 commit。异步项目中 commit 需要 await。

1
2
3
4
5
6
7
8
9
10
11
12
@app.put("/users/{user_id}")
async def update_user(user_id: int, db: AsyncSession = Depends(get_db)):
    user = await db.get(User_db, user_id)        # 先查询
    if not user:
        return {"error": "用户不存在"}

    user.username = "新用户名"                      # 修改属性
    user.email = "new@example.com"

    await db.commit()                              # 提交事务
    await db.refresh(user)                         # 刷新获取最新状态
    return user

配合 Pydantic 模型更新

实际开发中通常由请求体指定要修改的字段：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from pydantic import BaseModel

class UserUpdate(BaseModel):
    username: str | None = None     # 可选字段，不传则不修改
    email: str | None = None

@app.put("/users/{user_id}")
async def update_user(user_id: int, user_data: UserUpdate, db: AsyncSession = Depends(get_db)):
    user = await db.get(User_db, user_id)
    if not user:
        return {"error": "用户不存在"}

    # 只更新传入了的字段（非 None）
    update_data = user_data.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(user, field, value)

    await db.commit()
    await db.refresh(user)
    return user

model_dump(exclude_unset=True) 只返回请求体中实际传入了的字段，未传的字段为 None 但会被排除，从而实现部分更新。

批量更新

使用 update() 方法可以一次更新多条符合条件的数据，无需逐条查询，性能更高：

1
2
3
4
5
6
7
8
9
10
@app.put("/users")
async def batch_update_users(db: AsyncSession = Depends(get_db)):
    stmt = (
        update(User_db)
        .where(User_db.username.like("曹%"))
        .values(email="cao@example.com")
    )
    await db.execute(stmt)
    await db.commit()
    return {"message": "批量更新成功"}

使用 update() 方式更新不会触发 ORM 的 Python 层事件（如 @validates 装饰器），直接生成 SQL UPDATE 语句执行。如果需要事件回调，请使用先查后改的方式。

删除

删除单条数据

经典的三步：先查 → 调用 delete() → commit。

1
2
3
4
5
6
7
8
9
@app.delete("/users/{user_id}")
async def delete_user(user_id: int, db: AsyncSession = Depends(get_db)):
    user = await db.get(User_db, user_id)
    if not user:
        return {"error": "用户不存在"}

    await db.delete(user)        # 标记删除
    await db.commit()            # 提交事务，删除生效
    return {"message": "删除成功"}

db.delete() 只是将对象标记为待删除，真正执行 SQL 是在 commit() 时。

条件删除

使用 delete() 语句配合 where() 条件，直接执行 SQL 级别的删除，不经过 ORM 查询：

1
2
3
4
5
6
7
8
from sqlalchemy import delete

@app.delete("/users")
async def delete_users_by_condition(db: AsyncSession = Depends(get_db)):
    stmt = delete(User_db).where(User_db.username == "测试用户")
    await db.execute(stmt)
    await db.commit()
    return {"message": "条件删除成功"}

批量删除

delete() 配合 in_() 可以一次删除多条匹配的数据：

1
2
3
4
5
6
7
8
from sqlalchemy import delete

@app.delete("/users/batch")
async def batch_delete_users(ids: list[int], db: AsyncSession = Depends(get_db)):
    stmt = delete(User_db).where(User_db.id.in_(ids))
    result = await db.execute(stmt)
    await db.commit()
    return {"message": f"删除了 {result.rowcount} 条数据"}

result.rowcount 返回受影响的记录行数，可用于确认删除了多少条数据。

删除 vs 软删除

实际项目中往往不推荐物理删除（直接从数据库中抹去），更常见的是软删除——给表加一个标记字段来标识是否已”删除”：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from datetime import datetime
from sqlalchemy import DateTime

class User_db(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    username = Column(String)
    email = Column(String)
    is_deleted = Column(Boolean, default=False)          # 软删除标记
    deleted_at = Column(DateTime, nullable=True)          # 删除时间

# 软删除：更新标记，而非真正删除
@app.delete("/users/{user_id}")
async def soft_delete_user(user_id: int, db: AsyncSession = Depends(get_db)):
    user = await db.get(User_db, user_id)
    if not user:
        return {"error": "用户不存在"}

    user.is_deleted = True
    user.deleted_at = datetime.now()
    await db.commit()
    return {"message": "已软删除"}

# 查询时忽略已删除的数据
@app.get("/users")
async def get_user_list(db: AsyncSession = Depends(get_db)):
    stmt = select(User_db).where(User_db.is_deleted == False)
    res = await db.execute(stmt)
    users = res.scalars().all()
    return users

增删改查方法速查表

操作 方法 说明
新增单条 session.add(instance) 添加一个 ORM 实例到会话
新增多条 session.add_all([instances]) 批量添加 ORM 实例列表
查询全部 session.execute(select(Model)).scalars().all() 返回表中所有记录
主键查询 session.get(Model, pk) 根据主键快速查找
条件查询 select(Model).where(...) 链式添加筛选条件
聚合查询 func.max/min/avg/sum/count(Model.col) 配合 select 使用
分页查询 select(Model).offset(n).limit(m) 跳过 n 条取 m 条
更新属性 直接赋值 + session.commit() 修改实例属性后提交
批量更新 update(Model).where(...).values(...) SQL 级更新，不经过 ORM
删除单条 session.delete(instance) 标记删除，commit 后生效
批量删除 delete(Model).where(...) SQL 级删除
软删除 更新 is_deleted 标记字段 保留数据，仅标记不可见
提交事务 await session.commit() 异步提交，必须 await
刷新实例 await session.refresh(instance) 从数据库重新加载最新数据