新增与更新

Will大约 4 分钟

新增与更新

本节聚焦 ActiveModel 的插入/局部更新、批量插入、Upsert、乐观锁、时间戳与唯一约束处理。核心要点：

ActiveModel 字段通过 Set(...)/NotSet 标记写入与跳过
插入后返回插入的 Model（部分数据库会回填自增/默认值）
Update Many 支持条件批量更新
Postgres/MySQL 支持 Upsert（冲突时更新）
使用 ActiveModelBehavior 自动填充 created_at/updated_at
乐观锁基于 version 字段 + 条件更新实现

单条插入（ActiveModel）

use sea_orm::{ActiveModelTrait, Set};
use chrono::Utc;

let now = Utc::now();
let am = entity::sys_user::ActiveModel {
  id: Set(state.id_gen.next_id()),           // 若数据库自增，可 NotSet
  username: Set("alice".into()),
  password_hash: Set(hash_password("Secret@123")?),
  real_name: Set(Some("Alice".into())),
  dept_id: Set(None),
  status: Set(1),
  created_at: Set(now),
  updated_at: Set(now),
  deleted_at: Set(None),                     // 软删除字段（可选）
};

let model = am.insert(&state.db).await?;

提示

未 Set 的字段不会写入（若列非空/无默认，则会报错）
插入返回的 model 通常包含数据库默认值（依赖驱动与列定义）

局部更新（ActiveModel）

use sea_orm::{EntityTrait, ActiveModelTrait, Set};

let user = entity::sys_user::Entity::find_by_id(id).one(&db).await?
  .ok_or_else(|| anyhow::anyhow!("not found"))?;

let mut am = user.into_active_model();
am.real_name = Set(Some("Alice Zhang".into()));
am.status = Set(1);
am.updated_at = Set(chrono::Utc::now());

let updated = am.update(&db).await?;

仅对 Set 的列生成更新语句
未 Set 的列保持不变

批量插入（Insert Many）

use sea_orm::{ActiveModelTrait, Set, EntityTrait};

let now = chrono::Utc::now();
let items = (0..3).map(|i| entity::sys_user::ActiveModel {
  id: Set(state.id_gen.next_id()),
  username: Set(format!("user{i}")),
  password_hash: Set(hash_password("Pass@123")?),
  status: Set(1),
  created_at: Set(now),
  updated_at: Set(now),
  ..Default::default()
}).collect::<Vec<_>>();

let res = entity::sys_user::Entity::insert_many(items).exec(&db).await?;
let rows = res.rows_affected;

注意

insert_many 默认不返回每条记录（仅行数/主键起点与否视驱动）
批量插入量大时，建议分批（例如 100〜1000/批）

批量更新（Update Many）

use sea_orm::{EntityTrait, ColumnTrait, QueryFilter, Set};

let aff = entity::sys_user::Entity::update_many()
  .col_expr(entity::sys_user::Column::Status, 1.into())
  .filter(entity::sys_user::Column::DeptId.eq(1001))
  .exec(&db).await?;
let rows = aff.rows_affected;

col_expr 支持表达式：Expr::value(0)、Expr::col(...).add(1) 等

Upsert（冲突时更新）

PostgreSQL/MySQL 支持 ON CONFLICT / ON DUPLICATE KEY。使用 sea-query 构建器：

use sea_orm::{EntityTrait, DbBackend};
use sea_orm::sea_query::{Query, InsertStatement, OnConflict, Expr};

let builder = db.get_database_backend();
let mut stmt: InsertStatement = Query::insert()
  .into_table(entity::sys_user::Entity)
  .columns([
    entity::sys_user::Column::Id,
    entity::sys_user::Column::Username,
    entity::sys_user::Column::PasswordHash,
    entity::sys_user::Column::Status,
    entity::sys_user::Column::CreatedAt,
    entity::sys_user::Column::UpdatedAt,
  ])
  .values_panic([
    state.id_gen.next_id().into(),
    "alice".into(),
    hash_password("Secret@123")?.into(),
    1.into(),
    chrono::Utc::now().into(),
    chrono::Utc::now().into(),
  ])
  .on_conflict(
    OnConflict::column(entity::sys_user::Column::Username)
      .update_columns([
        entity::sys_user::Column::PasswordHash,
        entity::sys_user::Column::UpdatedAt,
      ])
      .to_owned(),
  )
  .to_owned();

let res = db.execute(builder.build(&db.get_database_backend())).await?;

Postgres：ON CONFLICT (username) DO UPDATE SET ...
MySQL：会生成 INSERT ... ON DUPLICATE KEY UPDATE ...

乐观锁（version 字段）

增加 version 整数列，每次更新时检查旧版本并自增：

use sea_orm::{EntityTrait, ColumnTrait, QueryFilter};

#[derive(sea_orm::FromQueryResult)]
struct Current { version: i32 }

let current: Option<Current> = entity::sys_user::Entity::find_by_id(id)
  .select_only()
  .column(entity::sys_user::Column::Version)
  .into_model()
  .one(&db).await?;
let old = current.ok_or_else(|| anyhow::anyhow!("not found"))?.version;

let aff = entity::sys_user::Entity::update_many()
  .col_expr(entity::sys_user::Column::RealName, "Alice V2".into())
  .col_expr(entity::sys_user::Column::Version, (old + 1).into())
  .filter(entity::sys_user::Column::Id.eq(id))
  .filter(entity::sys_user::Column::Version.eq(old))
  .exec(&db).await?;

anyhow::ensure!(aff.rows_affected == 1, "concurrent update detected");

自动时间戳（Behavior）

使用 ActiveModelBehavior 自动填充 created_at/updated_at：

// 在实体文件中
impl sea_orm::ActiveModelBehavior for ActiveModel {
  fn before_save(self, insert: bool) -> Result<Self, sea_orm::DbErr> {
    let mut m = self;
    let now = chrono::Utc::now();
    if insert {
      if m.created_at.is_not_set() { m.created_at = Set(now); }
    }
    m.updated_at = Set(now);
    Ok(m)
  }
}

唯一约束冲突与错误映射

将唯一键冲突（如用户名重复）映射为业务错误：

match active.insert(&db).await {
  Ok(m) => Ok(m),
  Err(sea_orm::DbErr::Exec(sea_orm::RuntimeErr::SqlxError(e)))
  | Err(sea_orm::DbErr::Query(sea_orm::RuntimeErr::SqlxError(e))) if is_unique_violation(&e) => {
    Err(AppError::conflict("用户名已存在").into())
  }
  Err(e) => Err(anyhow::anyhow!(e)),
}

is_unique_violation 可根据数据库错误码实现（PG 23505 / MySQL 1062）。

事务中的写操作

use sea_orm::{TransactionTrait, Set};

db.transaction(|txn| {
  Box::pin(async move {
    let now = chrono::Utc::now();
    let u = entity::sys_user::ActiveModel {
      id: Set(state.id_gen.next_id()),
      username: Set("bob".into()),
      password_hash: Set(hash_password("Pass@123")?),
      status: Set(1),
      created_at: Set(now),
      updated_at: Set(now),
      ..Default::default()
    }.insert(txn).await?;

    // 更多写操作...
    Ok::<_, anyhow::Error>(())
  })
}).await?;

小结

单条：ActiveModel insert/update
批量：insert_many/update_many
Upsert：使用 sea-query 的 OnConflict
乐观锁：version + 条件更新
自动时间戳：ActiveModelBehavior::before_save
唯一约束：捕获错误码映射业务错误