ecto-patterns

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Ecto Patterns for Phoenix/Elixir

适用于Phoenix/Elixir的Ecto模式

Ecto is the data layer for Phoenix applications: schemas, changesets, queries, migrations, and transactions. Good Ecto practice keeps domain logic in contexts, enforces constraints in the database, and uses transactions for multi-step workflows.

Ecto是Phoenix应用的数据层，负责Schema、Changeset、查询、迁移及事务管理。良好的Ecto实践要求将领域逻辑置于上下文（contexts）中，在数据库层面强制执行约束，并使用事务处理多步骤工作流。

Schemas and Changesets

Schema与Changeset

elixir

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

Guidelines

Keep casting/validation in changesets; keep business logic in contexts.
Always pair validation with DB constraints (
```
unique_constraint
```
,
```
foreign_key_constraint
```
).
Use
```
changeset/2
```
for updates; avoid mass assigning without casting.

elixir

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

实践准则

类型转换与校验逻辑放在Changeset中；业务逻辑置于上下文里。
始终将校验规则与数据库约束（
```
unique_constraint
```
、
```
foreign_key_constraint
```
）配对使用。
使用
```
changeset/2
```
处理更新操作；避免未经类型转换的批量赋值。

Migrations

数据库迁移

elixir

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

Safe migration tips

Prefer additive changes: add columns nullable, backfill, then enforce null: false.
For large tables: use
```
concurrently: true
```
for indexes; disable in
```
change
```
and wrap in
```
up/down
```
for Postgres.
Data migrations belong in separate modules called from
```
mix ecto.migrate
```
via
```
execute/1
```
or in distinct scripts; ensure idempotence.
Coordinate locks: avoid long transactions; break migrations into small steps.

elixir

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
```
中调用，或使用单独脚本；确保操作具有幂等性。
协调锁机制：避免长事务；将迁移拆分为多个小步骤。

Queries and Preloads

查询与预加载

elixir

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

Patterns

Use
```
preload
```
rather than calling Repo in loops; prefer
```
Repo.preload/2
```
after fetching.
Use
```
select
```
to avoid loading large blobs.
For concurrency, use
```
Repo.transaction
```
with
```
lock: "FOR UPDATE"
```
in queries that need row-level locks.

elixir

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
```
避免加载大对象数据。
并发场景下，在需要行级锁的查询中，结合
```
Repo.transaction
```
与
```
lock: "FOR UPDATE"
```
。

Transactions and Ecto.Multi

事务与Ecto.Multi

elixir

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

Guidelines

Prefer
```
Multi.run/3
```
for side effects that can fail; return
```
{:ok, value}
```
or
```
{:error, reason}
```
.
Use
```
Multi.update_all
```
for batch updates; include
```
where
```
guards to prevent unbounded writes.
Propagate errors upward; translate them in controllers/LiveViews.

elixir

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
```
条件防止无限制写入。
向上传递错误；在控制器或LiveViews中转换错误信息。

Associations and Constraints

关联与约束

Use
```
on_replace: :delete
```
/
```
:nilify
```
to control nested changes.
Define
```
foreign_key_constraint/3
```
and
```
unique_constraint/3
```
in changesets to surface DB errors cleanly.
For many-to-many, prefer join schema (
```
has_many :memberships
```
) instead of automatic
```
many_to_many
```
when you need metadata.

使用
```
on_replace: :delete
```
/
```
:nilify
```
控制嵌套变更的行为。
在Changeset中定义
```
foreign_key_constraint/3
```
与
```
unique_constraint/3
```
，以清晰呈现数据库错误。
多对多关联场景下，当需要元数据时，优先使用关联Schema（
```
has_many :memberships
```
）而非自动生成的
```
many_to_many
```
。

Pagination and Filtering

分页与过滤

Offset/limit for small datasets; cursor-based for large lists (
```
Scrivener
```
,
```
Flop
```
,
```
Paginator
```
).
Normalize filters in contexts; avoid letting controllers build queries directly.
Add composite indexes to match filter columns; verify with
```
EXPLAIN ANALYZE
```
.

小数据集使用偏移量/限制（offset/limit）；大型列表使用基于游标分页（可借助
```
Scrivener
```
、
```
Flop
```
、
```
Paginator
```
工具）。
在上下文中统一处理过滤逻辑；避免让控制器直接构建查询语句。
为过滤字段添加复合索引；使用
```
EXPLAIN ANALYZE
```
验证索引效果。

Multi-Tenancy Patterns

多租户模式

Prefix-based: Postgres schemas per tenant (
```
put_source/2
```
with
```
prefix:
```
) — good isolation, needs per-tenant migrations.
Row-based:
```
tenant_id
```
column + row filters — simpler migrations; add partial indexes per tenant when large.
Always scope queries by tenant in contexts; consider using policies/guards to enforce.

基于前缀：为每个租户创建独立的Postgres Schema（通过
```
put_source/2
```
指定
```
prefix:
```
参数）——隔离性好，但需要为每个租户执行迁移。
基于行：通过
```
tenant_id
```
列+行级过滤实现——迁移更简单；数据量大时可为每个租户添加部分索引。
始终在上下文中按租户范围限定查询；可考虑使用策略/守卫机制强制执行。

Performance and Ops

性能与运维

Use
```
Repo.stream
```
for large exports; wrap in
```
Repo.transaction
```
.
Cache hot reads with ETS/Cachex; invalidate on writes.
Watch query counts in LiveView/Channels; preload before rendering to avoid N+1.
Telemetry:
```
OpentelemetryEcto
```
exports query timings; add DB connection pool metrics.

导出大量数据时使用
```
Repo.stream
```
；需包裹在
```
Repo.transaction
```
中执行。
使用ETS/Cachex缓存高频读操作；写入时失效缓存。
监控LiveView/Channels中的查询次数；渲染前预加载关联数据以避免N+1查询问题。
遥测：
```
OpentelemetryEcto
```
可导出查询耗时数据；添加数据库连接池指标监控。

Testing

测试

elixir

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
```
sets up sandboxed DB; keep tests async unless transactions conflict.
Use factories/fixtures in
```
test/support
```
to build valid structs quickly.
For migrations, add regression tests for constraints (unique/index-backed constraints).

elixir

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
```
中使用工厂/固定数据快速构建合法结构体。
针对迁移，为约束（唯一/索引约束）添加回归测试。

Common Pitfalls

常见陷阱

Running risky DDL in a single migration step (avoid locks; break apart).
Skipping DB constraints and relying only on changesets.
Querying associations in loops instead of preloading.
Missing transactions for multi-step writes (partial state on failure).
Forgetting tenant scoping on read/write in multi-tenant setups.

在单个迁移步骤中执行高风险DDL操作（避免锁；拆分步骤）。
仅依赖Changeset而忽略数据库约束。
循环查询关联数据而非使用预加载。
多步骤写入操作未使用事务（失败时导致数据状态不一致）。
多租户场景下读写操作未按租户范围限定。