https://www.viget.com/articles/multi-tenancy-with-postgres-schemas-and-ecto/

Repo.ex

defmodule Acme.Repo do
  use Ecto.Repo,
    otp_app: :acme,
    adapter: Ecto.Adapters.Postgres

  def with_prefix(prefix) do
    module_atom = Module.concat([Acme, Repo, WithPrefix, Macro.camelize(prefix)])

    # We could not find a better way to see if this module already existed
    if !Kernel.function_exported?(module_atom, :prefix, 0) do
      Module.create(module_atom, module_contents(prefix), Macro.Env.location(__ENV__))
    end

    module_atom
  end

  def get_prefix(%{__struct__: _} = struct) do
    ["tenant" | tail] =
      struct
      |> Ecto.get_meta(:prefix)
      |> String.split("_")

    tail
    |> Enum.join("_")
  end

  defp module_contents(prefix) do
    quote do
      def prefix(), do: unquote(prefix)

      def schema_prefix(), do: "tenant_" <> unquote(prefix)

      defp opts_with_prefix(opts \\ []) do
        Keyword.put_new(opts, :prefix, schema_prefix())
      end

      def paginate(%Ecto.Query{} = queryable, params) do
        queryable
        |> Map.put(:prefix, schema_prefix())
        |> Acme.Repo.paginate(params)
      end

      def paginate(data, params) do
        data
        |> Ecto.Queryable.to_query()
        |> paginate(params)
      end

      defdelegate __adapter__(), to: Acme.Repo

      def aggregate(queryable, aggregate, opts \\ [])

      def aggregate(queryable, aggregate, opts) when is_list(opts) do
        Acme.Repo.aggregate(queryable, aggregate, opts_with_prefix(opts))
      end

      def aggregate(queryable, aggregate, field) when is_atom(field) do
        Acme.Repo.aggregate(queryable, aggregate, field, opts_with_prefix())
      end

      def aggregate(queryable, aggregate, field, opts)
          when is_atom(field) and is_list(opts) do
        Acme.Repo.aggregate(queryable, aggregate, field, opts_with_prefix(opts))
      end

      def all(queryable, opts \\ []) do
        Acme.Repo.all(queryable, opts_with_prefix(opts))
      end

      defdelegate checkout(function, opts), to: Acme.Repo

      defdelegate config(), to: Acme.Repo

      defdelegate default_options(operation), to: Acme.Repo

      def delete(struct_or_changeset, opts \\ []) do
        Acme.Repo.delete(struct_or_changeset, opts_with_prefix(opts))
      end

      def delete!(struct_or_changeset, opts \\ []) do
        Acme.Repo.delete!(struct_or_changeset, opts_with_prefix(opts))
      end

      def delete_all(queryable, opts \\ []) do
        Acme.Repo.delete_all(queryable, opts_with_prefix(opts))
      end

      def exists?(queryable, opts \\ []) do
        Acme.Repo.exists?(queryable, opts_with_prefix(opts))
      end

      def get(queryable, id, opts \\ []) do
        Acme.Repo.get(queryable, id, opts_with_prefix(opts))
      end

      def get!(queryable, id, opts \\ []) do
        Acme.Repo.get!(queryable, id, opts_with_prefix(opts))
      end

      def get_by(queryable, clauses, opts \\ []) do
        Acme.Repo.get_by(queryable, clauses, opts_with_prefix(opts))
      end

      def get_by!(queryable, clauses, opts \\ []) do
        Acme.Repo.get_by!(queryable, clauses, opts_with_prefix(opts))
      end

      defdelegate get_dynamic_repo(), to: Acme.Repo

      defdelegate in_transaction?(), to: Acme.Repo

      defdelegate init(context, config), to: Acme.Repo

      def insert(struct_or_changeset, opts \\ []) do
        Acme.Repo.insert(struct_or_changeset, opts_with_prefix(opts))
      end

      def insert!(struct_or_changeset, opts \\ []) do
        Acme.Repo.insert!(struct_or_changeset, opts_with_prefix(opts))
      end

      def insert_all(schema_or_source, entries, opts \\ []) do
        Acme.Repo.insert_all(schema_or_source, entries, opts_with_prefix(opts))
      end

      def insert_or_update(changeset, opts \\ []) do
        Acme.Repo.insert_or_update(changeset, opts_with_prefix(opts))
      end

      def insert_or_update!(changeset, opts \\ []) do
        Acme.Repo.insert_or_update!(changeset, opts_with_prefix(opts))
      end

      defdelegate load(module_or_map, data), to: Acme.Repo

      def one(queryable, opts \\ []) do
        Acme.Repo.one(queryable, opts_with_prefix(opts))
      end

      def one!(queryable, opts \\ []) do
        Acme.Repo.one!(queryable, opts_with_prefix(opts))
      end

      def preload(queryable, preloads, opts \\ []) do
        Acme.Repo.preload(queryable, preloads, opts_with_prefix(opts))
      end

      defdelegate prepare_query(operation, query, opts \\ []), to: Acme.Repo

      defdelegate put_dynamic_repo(arg1), to: Acme.Repo

      defdelegate rollback(value), to: Acme.Repo

      defdelegate start_link(opts \\ []), to: Acme.Repo

      defdelegate stop(opts \\ []), to: Acme.Repo

      def stream(queryable, opts \\ []) do
        Acme.Repo.stream(queryable, opts_with_prefix(opts))
      end

      def transaction(fun_or_multi, opts \\ []) do
        Ecto.Repo.Transaction.transaction(__MODULE__, Acme.Repo, fun_or_multi, opts)
      end

      def update(changeset, opts \\ []) do
        Acme.Repo.update(changeset, opts_with_prefix(opts))
      end

      def update!(changeset, opts \\ []) do
        Acme.Repo.update!(changeset, opts_with_prefix(opts))
      end

      def update_all(queryable, updates, opts \\ []) do
        Acme.Repo.update_all(queryable, updates, opts_with_prefix(opts))
      end
    end
  end
end