Elixir Ecto 数据访问层：Schema、Query、迁移与变更集

# config/config.exs
config :my_app, MyApp.Repo,
  username: "postgres",
  password: "postgres",
  hostname: "localhost",
  database: "my_app_dev",
  pool_size: 10,
  log: false

# config/runtime.exs（生产环境用环境变量）
config :my_app, MyApp.Repo,
  url: System.get_env("DATABASE_URL"),
  pool_size: String.to_integer(System.get_env("POOL_SIZE", "10"))

# lib/my_app/repo.ex
defmodule MyApp.Repo do
  use Ecto.Repo,
    otp_app: :my_app,
    adapter: Ecto.Adapters.Postgres
end

# 单记录操作
user = Repo.get(User, 42)
user = Repo.get!(User, 42)                     # 不存在抛异常
user = Repo.get_by(User, email: "a@ex.com")
user = Repo.one!(from u in User, where: u.id == 42)

# 写入
{:ok, user} = Repo.insert(%User{name: "Alice"})
{:ok, user} = Repo.update(user)
{:ok, _} = Repo.delete(user)
{:ok, user} = Repo.insert_or_update(changeset)

# 批量
Repo.insert_all(User, [%{name: "A"}, %{name: "B"}], on_conflict: :nothing)
Repo.delete_all(User)

# 聚合
Repo.aggregate(User, :count, :id)

# 事务
Repo.transaction(fn ->
  Repo.insert!(changeset1)
  Repo.insert!(changeset2)
end)

defmodule MyApp.Blog.Post do
  use Ecto.Schema
  import Ecto.Changeset

  schema "posts" do
    field :title, :string
    field :body, :string
    field :published_at, :utc_datetime
    field :view_count, :integer, default: 0
    field :rating, :float, default: 0.0
    field :meta, :map, default: %{}
    field :tags, {:array, :string}, default: []
    field :status, Ecto.Enum, values: [:draft, :published, :archived], default: :draft

    belongs_to :author, MyApp.Accounts.User
    has_many :comments, MyApp.Blog.Comment
    has_one :seo_meta, MyApp.Blog.SeoMeta

    timestamps()   # inserted_at, updated_at
  end
end

schema "products" do
  field :price_cents, :integer
  field :currency, :string

  # 虚拟字段：不落库，仅用于表单与计算
  field :display_price, :string, virtual: true
  field :confirm, :boolean, virtual: true
end

def changeset(product, attrs) do
  product
  |> cast(attrs, [:price_cents, :currency, :confirm])
  |> validate_required([:price_cents, :currency])
  |> validate_confirmation(:price_cents)     # price_cents_confirmation 一致
  |> put_calc(:display_price)
end

defp put_calc(changeset) do
  case changeset do
    %Ecto.Changeset{valid?: true, changes: %{price_cents: c, currency: cur}} ->
      put_change(changeset, :display_price, "#{c / 100} #{cur}")
    _ ->
      changeset
  end
end

import Ecto.Query

# 全表 + 排序 + 限制
posts = Repo.all(
  from p in Post,
    where: p.published_at != nil,
    order_by: [desc: p.published_at],
    limit: 20
)

# 管道风格
query =
  Post
  |> where([p], p.view_count > 1000)
  |> order_by([p], desc: p.view_count)
  |> limit(10)

Repo.all(query)

# 动态条件（参数插值，防注入）
title = "Ecto"
Repo.all(from p in Post, where: p.title == ^title)

from p in Post,
  where: p.title like "%Ecto%",
  where: p.published_at > ^Date.shift(Date.utc_today(), day: -30),
  where: p.view_count in [100, 200, 300],
  where: p.status == :published or p.status == :archived,
  where: fragment("lower(?) LIKE ?", p.title, "%ecto%"),   # 原生 SQL 片段
  select: %{title: p.title, views: p.view_count}

# 按状态统计
Repo.all(
  from p in Post,
    group_by: p.status,
    select: {p.status, count(p.id)}
)

# having 过滤
Repo.all(
  from p in Post,
    group_by: p.status,
    having: count(p.id) > 10,
    select: {p.status, count(p.id)}
)

# 窗口函数（PG 支持）
Repo.all(
  from p in Post,
    select: %{title: p.title, row_num: row_number() |> over(partition_by: p.status, order_by: p.view_count)}
)

# join 查询（Ecto 自动按外键匹配）
query =
  from p in Post,
    join: c in assoc(p, :comments),
    where: c.approved == true,
    distinct: true,
    select: p

# 预加载（避免 N+1）
posts = Repo.all(from p in Post, preload: [:author, :comments])

# 嵌套预加载 + 过滤
posts = Repo.all(
  from p in Post,
    preload: [author: :profile, comments: {c in Comment, where: c.approved == true}]
)

# 批量预加载
posts = Repo.all(Post)
Repo.preload(posts, :comments)

# 子查询：热门作者下的最新文章
popular_authors =
  from u in User,
    join: p in assoc(u, :posts),
    group_by: u.id,
    having: count(p.id) > 10,
    select: u.id

query =
  from p in Post,
    where: p.author_id in subquery(popular_authors),
    select: p

# CTE（with_cte，PG 支持）
cte_query =
  from p in Post,
    where: p.published_at != nil,
    order_by: [desc: p.published_at],
    limit: 10

from p in Post,
  with_cte: "latest_posts", as: cte_query,
  join: lp in "latest_posts", on: lp.id == p.id,
  select: p

# 生成迁移文件
mix ecto.gen.migration create_posts

# 生成带结构迁移（跟随 schema）
mix ecto.gen.migration add_view_count_to_posts

# 执行
mix ecto.migrate
# 回滚一步
mix ecto.rollback
# 回滚到指定版本
mix ecto.rollback --to 20240926120000

# priv/repo/migrations/20260926010000_create_posts.exs
defmodule MyApp.Repo.Migrations.CreatePosts do
  use Ecto.Migration

  def change do
    create table(:posts) do
      add :title, :string, null: false, size: 200
      add :body, :text
      add :status, :string, default: "draft", null: false
      add :view_count, :integer, default: 0, null: false
      add :author_id, references(:users, on_delete: :delete_all), null: false

      timestamps()
    end

    create index(:posts, [:author_id])
    create index(:posts, [:status, :view_count])

    # 唯一约束（用于幂等）
    create unique_index(:posts, [:title])
  end
end

def change do
  # change/0 自动推导 up/down
  create table(:posts) do
    add :title, :string
  end
  create index(:posts, [:title])
end

# 不可逆操作需分开 up/down
def up do
  execute "CREATE EXTENSION pg_trgm"
end

def down do
  execute "DROP EXTENSION pg_trgm"
end

# 数据迁移（谨慎）
def up do
  execute "UPDATE posts SET status = 'published' WHERE published_at IS NOT NULL"
end

def changeset(post, attrs) do
  post
  |> cast(attrs, [:title, :body, :status])   # 白名单字段 + 类型转换
  |> validate_required([:title])
  |> validate_length(:title, min: 3, max: 200)
  |> validate_inclusion(:status, [:draft, :published, :archived])
  |> validate_number(:view_count, greater_than_or_equal_to: 0)
  |> validate_format(:title, ~r/^[A-Za-z0-9\s]+$/, message: "仅允许字母数字")
  |> unique_constraint(:title)                # 依赖数据库唯一索引
  |> assoc_constraint(:author)                # 外键必须存在
  |> check_constraint(:status, name: :status_check)
end

def changeset(post, attrs) do
  post
  |> cast(attrs, [:title, :body])
  |> validate_required([:title])
  |> validate_body_not_duplicate()
end

defp validate_body_not_duplicate(changeset) do
  case changeset do
    %Ecto.Changeset{changes: %{body: body}, valid?: true} ->
      case MyApp.Blog.find_duplicate(body) do
        nil -> changeset
        _post -> add_error(changeset, :body, "已存在相同内容")
      end
    _ -> changeset
  end
end

def changeset(post, attrs) do
  post
  |> cast(attrs, [:title, :author_id])
  |> validate_required([:title, :author_id])
  |> unique_constraint(:title, message: "标题已存在")
  |> assoc_constraint(:author)
end

# 约束冲突时，Ecto 捕获数据库错误转为 changeset error：
#   - unique_constraint  → :unique 约束错误
#   - assoc_constraint   → :foreign_key 约束错误
#   - check_constraint   → :check 约束错误
#   - exclusion_constraint → 排除约束（如 PG 排他索引）

# 嵌套关联：创建 Post 时同时创建 Comments
def changeset(post, attrs) do
  post
  |> cast(attrs, [:title, :body])
  |> cast_assoc(:comments)              # 需要 schema 中 has_many :comments
  |> validate_required([:title])
end

# 显式操作 has_many
def put_comments(changeset, comments) do
  put_assoc(changeset, :comments, comments)
end

# 单独更新关联（不触碰父记录）
comment_changeset =
  Comment.changeset(comment, attrs)
  |> Repo.update()

# 从 map 构建
changeset = Ecto.Changeset.cast(%Post{}, attrs, [:title])

# 合并 / 附加
changeset
|> Ecto.Changeset.put_change(:slug, slugify(changeset.changes.title))
|> Ecto.Changeset.put_new(:view_count, 0)   # 仅在未设值时写入
|> Ecto.Changeset.force_change(:status, :draft)

# 变更集状态检查
changeset.valid?
changeset.errors      # [{field, {msg, opts}}]
changeset.changes     # 只有被变更的字段
Ecto.Changeset.traverse_errors(changeset, &format/2)   # 格式化错误