ecto-patterns

Phoenix/Elixir 的 Ecto 模式

Ecto 是 Phoenix 应用的数据层，负责架构、变更集、查询、迁移和事务。良好的 Ecto 实践会将领域逻辑保留在上下文中，在数据库中强制约束，并使用事务处理多步骤工作流。

架构和变更集

defmodule MyApp.Accounts.User do
  use Ecto.Schema
  import Ecto.Changeset

  schema "users" do
    field :email, :string
    field :hashed_password, :string
    field :confirmed_at, :naive_datetime
    has_many :memberships, MyApp.Orgs.Membership
    timestamps()
  end

  def registration_changeset(user, attrs) do
    user
    |> cast(attrs, [:email, :password])
    |> validate_required([:email, :password])
    |> validate_format(:email, ~r/@/)
    |> validate_length(:password, min: 12)
    |> unique_constraint(:email)
    |> hash_password()
  end

  defp hash_password(%{valid?: true} = cs),
    do: put_change(cs, :hashed_password, Argon2.hash_pwd_salt(get_change(cs, :password)))
  defp hash_password(cs), do: cs
end

指南

• 在变更集中进行类型转换和验证；在上下文中处理业务逻辑。
• 始终将验证与数据库约束配对使用（unique_constraint、foreign_key_constraint）。
• 更新时使用 changeset/2；避免不经类型转换就批量赋值。

迁移

def change do
  create table(:users) do
    add :email, :citext, null: false
    add :hashed_password, :string, null: false
    add :confirmed_at, :naive_datetime
    timestamps()
  end

  create unique_index(:users, [:email])
end

安全迁移提示

• 优先采用增量变更：先添加可空列，回填数据，然后强制设为 null: false。
• 对于大型表：为索引使用 concurrently: true；在 Postgres 中，将其从 change 中禁用，并用 up/down 包装。
• 数据迁移应放在单独的模块中，通过 execute/1 从 mix ecto.migrate 调用，或放在独立脚本中；确保操作幂等。
• 协调锁：避免长事务；将迁移拆分为多个小步骤。

查询和预加载

import Ecto.Query

def list_users(opts \\ %{}) do
  base =
    from u in MyApp.Accounts.User,
      preload: [:memberships],
      order_by: [desc: u.inserted_at]

  Repo.all(apply_pagination(base, opts))
end

defp apply_pagination(query, %{limit: limit, offset: offset}),
  do: query |> limit(^limit) |> offset(^offset)
defp apply_pagination(query, _), do: query

模式

• 使用 preload，而不是在循环中调用 Repo；优先在获取数据后调用 Repo.preload/2。
• 使用 select 避免加载大型二进制数据。
• 对于并发操作，在需要行级锁的查询中使用带有 lock: "FOR UPDATE" 的 Repo.transaction。

事务和 Ecto.Multi

alias Ecto.Multi

def onboard_user(attrs) do
  Multi.new()
  |> Multi.insert(:user, User.registration_changeset(%User{}, attrs))
  |> Multi.insert(:org, fn %{user: user} ->
    Org.changeset(%Org{}, %{owner_id: user.id, name: attrs["org_name"]})
  end)
  |> Multi.run(:welcome, fn _repo, %{user: user} ->
    MyApp.Mailer.deliver_welcome(user)
    {:ok, :sent}
  end)
  |> Repo.transaction()
end

指南

• 对于可能失败的副作用，优先使用 Multi.run/3；返回 {:ok, value} 或 {:error, reason}。
• 使用 Multi.update_all 批量更新；添加 where 条件，防止无界写入。
• 向上层传递错误；在控制器/LiveView 中转换错误。

关联和约束

• 使用 on_replace: :delete/:nilify 控制嵌套变更。
• 在变更集中定义 foreign_key_constraint/3 和 unique_constraint/3，以便清晰地呈现数据库错误。
• 对于多对多关联，如果需要存储元数据，优先使用联接架构（has_many :memberships），而不是自动的 many_to_many。

分页和筛选

• 小型数据集使用 Offset/limit；大型列表使用基于游标的分页（Scrivener、Flop、Paginator）。
• 在上下文中规范化筛选条件；避免让控制器直接构建查询。
• 添加与筛选列相匹配的复合索引；使用 EXPLAIN ANALYZE 进行验证。

多租户模式

• 基于前缀：每个租户使用独立的 Postgres 架构（通过带有 prefix: 的 put_source/2）——隔离性好，但需要为每个租户执行迁移。
• 基于行：使用 tenant_id 列和行筛选——迁移更简单；数据量较大时，为每个租户添加部分索引。
• 始终在上下文中按租户限定查询；考虑使用策略/守卫来强制执行。

性能和运维

• 对大型导出使用 Repo.stream；并在 Repo.transaction 中执行。
• 使用 ETS/Cachex 缓存热点读取；在写入时使缓存失效。
• 关注 LiveView/Channels 中的查询次数；在渲染前预加载，以避免 N+1 查询。
• 遥测：OpentelemetryEcto 会导出查询耗时；添加数据库连接池指标。

测试

use MyApp.DataCase, async: true

test "registration changeset validates email" do
  changeset = User.registration_changeset(%User{}, %{email: "bad", password: "secretsecret"})
  refute changeset.valid?
  assert %{email: ["has invalid format"]} = errors_on(changeset)
end

• DataCase 会设置沙盒数据库；除非事务冲突，否则保持测试异步执行。
• 在 test/support 中使用工厂/固定数据，快速构建有效的结构体。
• 对于迁移，为约束（唯一约束/索引支持的约束）添加回归测试。

常见陷阱

• 在单个迁移步骤中执行有风险的 DDL（避免锁；拆分迁移）。
• 跳过数据库约束，只依赖变更集。
• 在循环中查询关联，而不是预加载。
• 多步骤写入缺少事务（失败时会留下部分状态）。
• 在多租户设置中，读写时忘记按租户限定。