step = Ecto.get(from(u in User, where: u.email == ^email)) If the record is not found, returns :error, %Uni.Errorreason: :not_found . step = Ecto.list(MyApp.User) # Or with a query step = Ecto.list(from(u in User, where: u.active == true)) 6. run/2 – Raw Ecto Function For operations not covered, use run :
Uni.new() |> Ecto.get(MyApp.User, user_id) |> Uni.if( fn ctx -> ctx.data.get_user.active? end, do: Ecto.update(change_active_status_changeset(ctx.data.get_user)), else: Ecto.run(fn _repo -> :error, :user_inactive end) ) Uni provides Uni.on_error/2 to rescue failed steps:
Or pass a query with preloads directly to get/3 or list/3 . Sometimes you only want to insert if a condition is true: uni ecto plugin
alias Uni.Ecto step = Ecto.get(MyApp.Post, post_id) |> Ecto.preload([:comments, :author])
def registration_changeset(attrs) do %{} |> cast(attrs, [:email, :name]) |> validate_required([:email, :name]) |> validate_format(:email, ~r/@/) |> unique_constraint(:email) end end Step 2: Define the Pipeline Using Uni # lib/my_app/accounts/user_registration.ex defmodule MyApp.Accounts.UserRegistration do use Uni.Step import Uni.Ecto def run(attrs) do Uni.new() |> Uni.put(:original_attrs, attrs) |> add_step(:build_changeset, fn ctx -> changeset = User.registration_changeset(ctx.data.original_attrs) :ok, changeset end) |> add_step(:insert_user, insert(&1.data.build_changeset)) |> add_step(:assign_default_role, fn ctx -> user = ctx.data.insert_user updated_changeset = User.changeset(user, %role: "verified_member") :ok, updated_changeset end) |> add_step(:update_role, update(&1.data.assign_default_role)) |> add_step(:send_welcome, fn ctx -> # Imagine this calls an email service IO.inspect("Sending welcome to #ctx.data.update_role.email") :ok, %delivered: true end) |> Uni.execute() end end Step 3: Execute case UserRegistration.run(%email: "alice@example.com", name: "Alice") do :ok, %update_role: user -> IO.inspect(user, label: "Registered user") :error, step_name, error, _ctx -> IO.puts("Failed at step #step_name: #inspect(error)") end Key observation: Each step’s result is automatically stored under its step name ( :insert_user , :update_role ) inside the pipeline context. The named access ( ctx.data.insert_user ) makes dependencies explicit and traceable. Part 6: Advanced Usage Patterns Pattern 1: Transactions with Ecto.transaction/2 Uni can wrap a set of steps in a database transaction. Use the transaction/2 step: step = Ecto
defp deps do [ :ecto_sql, "~> 3.0", :uni, "~> 1.0", :uni_ecto, "~> 0.5" # The Uni Ecto Plugin ] end Run mix deps.get . The plugin works with any Ecto repository. You must explicitly tell Uni which repo to use as the default . This can be set in your config/config.exs :
def transfer_funds(from_id, to_id, amount) do Uni.new() |> Ecto.transaction(fn -> Uni.new() |> add_step(:decrement, Debit.run(from_id, amount)) |> add_step(:increment, Credit.run(to_id, amount)) |> Uni.execute() end) |> Uni.execute() end If :decrement fails, :increment never runs, and the transaction rolls back. Ecto’s preloading is crucial. Use Ecto.preload/3 as a step: end, do: Ecto
# test_helper.exs Uni.Ecto.Test.start(repo: MyApp.Repo, mode: :no_db) Now Ecto.get/3 returns a predefined fixture. This enables lightning-fast unit tests. | Feature | Raw Ecto | With Contexts (no Uni) | Uni + Ecto Plugin | |---------|----------|------------------------|--------------------| | Error tracking | :error, term | :error, term | Structured Uni.Error , step name included | | Transactions | Manual Repo.transaction/1 | Nested callbacks | Declarative Ecto.transaction/2 | | Side effects | Interleaved | Mixed in functions | Separate steps, auto-halt on error | | Testability | Mox or sandbox | Partial mocks | Per-step stubs + telemetry | | Readability | with chains | Varies | Linear pipeline with named steps | Part 9: Real-World Example – Blog Post with Tags Let’s finish with a non-trivial example: creating a blog post with tags, ensuring tags are upserted (find or create), and linking.