如何从Postgres拉取Schema更新至Elixir应用模型层？

一、替换原生工具解决类型支持与增量更新问题 #

原生mix ecto.gen.schema对枚举、jsonb的处理不完善，也不支持增量更新，推荐用以下方式处理：

1. 用ecto_autoschema生成/更新Schema #

这个工具专门针对现有数据库反向生成Ecto Schema，支持增量更新和复杂类型处理：

• 先在mix.exs中添加依赖：

  defp deps do
    [
      {:ecto_autoschema, "~> 0.5", only: :dev}
    ]
  end
  

• 安装依赖：

  mix deps.get
  

• 执行生成/更新命令（适配你的多Schema配置）：

  mix ecto_autoschema.gen --repo RealtimeService.Repo.Public --schema public --output lib/realtime_service/db/public/models --module RealtimeService.Db.Public.Models
  

  它会自动识别Postgres枚举类型并生成对应的Elixir枚举模块，把jsonb字段映射为map()类型，而且只会更新变更的字段——不会覆盖你在Schema文件中添加的自定义逻辑（比如关联、验证、业务方法）。

2. 自定义Mix任务（按需精细控制） #

如果需要更灵活的处理逻辑，可以写一个自定义Mix任务：

defmodule Mix.Tasks.Realtime.Sync.Schema do
  use Mix.Task

  def run(_args) do
    {:ok, _pid} = RealtimeService.Repo.Public.start_link()
    # 读取public Schema下的所有表
    tables = Ecto.Adapters.Postgres.list_tables(RealtimeService.Repo.Public, schema: "public")
    Enum.each(tables, &sync_table_schema/1)
  end

  defp sync_table_schema(table_name) do
    # 查询表字段详情
    columns = Ecto.Adapters.Postgres.query!(RealtimeService.Repo.Public, """
      SELECT column_name, data_type, udt_name
      FROM information_schema.columns
      WHERE table_schema = 'public' AND table_name = $1
      ORDER BY ordinal_position
    """, [table_name]).rows

    # 处理字段类型：枚举转Elixir Enum，jsonb转map()
    fields = Enum.map(columns, fn col ->
      {col_name, data_type, udt_name} = col
      type = case data_type do
        "jsonb" -> "map()"
        "ARRAY" -> "#{elem(data_type, 1)}[]"
        _ -> if String.starts_with?(udt_name, "enum_"), do: "Ecto.Enum, values: #{Enum.map(get_enum_values(udt_name), &inspect/1)}", else: data_type
      end
      "#{col_name}: :#{type}"
    end)

    # 生成/更新Schema文件，仅替换字段部分，保留自定义代码
    schema_file = Path.join(["lib", "realtime_service", "db", "public", "models", "#{table_name}.ex"])
    if File.exists?(schema_file) do
      content = File.read!(schema_file)
      # 替换schema块内的字段定义
      new_content = Regex.replace(~r/schema ".*?" do\n(.*?)\nend/s, content, fn _match, _old_fields ->
        "schema \"#{table_name}\" do\n  #{Enum.join(fields, "\n  ")}\nend"
      end)
      File.write!(schema_file, new_content)
    else
      # 生成新的Schema文件
      module_name = Macro.camelize(table_name)
      content = """
      defmodule RealtimeService.Db.Public.Models.#{module_name} do
        use Ecto.Schema
        import Ecto.Changeset

        schema "#{table_name}" do
          #{Enum.join(fields, "\n  ")}
        end

        @doc false
        def changeset(#{table_name}, attrs) do
          #{table_name}
          |> cast(attrs, [#{Enum.map(columns, fn {name, _, _} -> "\"#{name}\"" end) |> Enum.join(", ")}])
          |> validate_required([#{Enum.map(columns, fn {name, _, _} -> "\"#{name}\"" end) |> Enum.join(", ")}])
        end
      end
      """
      File.write!(schema_file, content)
    end
  end

  defp get_enum_values(enum_name) do
    Ecto.Adapters.Postgres.query!(RealtimeService.Repo.Public, """
      SELECT unnest(enum_range(NULL::#{enum_name}))
    """, []).rows |> Enum.map(&hd/1)
  end
end

运行这个任务即可同步：

mix realtime.sync.schema

二、固化Node作为唯一Schema变更源的流程 #

• 所有Schema变更（包括Elixir专属的elx Schema）必须通过Node应用的迁移工具执行（比如TypeORM、Prisma），禁止Elixir直接修改数据库结构
• 给Elixir的数据库账号设置public Schema只读权限，从数据库层面防止误修改：

  REVOKE ALL ON SCHEMA public FROM elixir_realtime_user;
  GRANT USAGE, SELECT ON ALL TABLES IN SCHEMA public TO elixir_realtime_user;
  ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO elixir_realtime_user;
  

• Elixir仅在需要更新模型层时，运行上述生成工具拉取最新结构

三、确保增量更新不覆盖自定义代码 #

• 使用ecto_autoschema时，它会自动保留你在Schema文件中添加的自定义逻辑（如关联、额外验证、业务方法），仅更新字段定义
• 自定义脚本中通过正则匹配替换schema/2块内的字段部分，不修改其他代码
• 建议将生成的Schema文件纳入版本控制，每次更新后对比差异，确保自定义代码未被覆盖

内容的提问来源于stack exchange，提问作者Abhinav Kulshreshtha