Basic CRUD

A lode port of Ecto’s Basic CRUD cheatsheet. Throughout this page, “internal data” means values your own Gleam code produces (already trusted and typed), while “external data” means input arriving from forms, APIs, or the CLI — which should pass through lode/changeset to be pruned and validated before it touches the database.

Three conventions differ from Elixir’s Ecto by design (see Divergences from Ecto):

The repo is a value, not a configured module — build it once and pass it explicitly.
Results, not exceptions — every operation returns Result(_, LodeError); there are no ! variants.
No macros — queries and changesets are ordinary values built with functions and pipes; labelled arguments replace keyword options.

Entries Ecto has that lode cannot express yet are kept below with a [TASK:...] marker.

Setup

Where Ecto configures a Repo module in config/config.exs, lode has no global configuration: open a pog connection pool and wrap it in a repo value. The long version (schemas, field accessors, codegen) lives in Getting Started.

# gleam.toml — until the packages are renamed and published to Hex,
# depend on a checkout by path (see README "Publishing").
[dependencies]
lode = { path = "../lode/lode" }         # changesets, query builder, repo, multi
lode_sql = { path = "../lode/lode_sql" } # Postgres adapter + migrations
pog = ">= 4.1.0 and < 5.0.0"

import lode/adapters/postgres
import lode/repo
import gleam/erlang/process
import pog

pub fn connect() -> repo.Repo {
  let assert Ok(started) =
    pog.default_config(process.new_name("db"))
    |> pog.host("localhost")
    |> pog.database("my_app_dev")
    |> pog.rows_as_map(True)
    |> pog.pool_size(10)
    |> pog.start
  repo.new(postgres.new(started.data))
}

For tests, swap the adapter and nothing else changes: repo.new(memory.new()) (from lode/adapters/memory).

The examples below assume r is a Repo plus the schemas and typed field accessors built in Getting Started:

pub type Movie { Movie(id: Int, title: String, tagline: String) }
pub type Person { Person(id: Int, name: String, age: Int) }

fn movie_schema() -> Schema(Movie) { ... }  // schema.new(source: "movies", ...)
fn person_schema() -> Schema(Person) { ... }

// Typed accessors make query expressions compile-checked
// (codegen emits these for you):
fn movie_id() -> FieldRef(Movie, Int) {
  expr.FieldRef(binding: 0, name: "id", lode_type: primitive.id())
}
fn movie_title() -> FieldRef(Movie, String) {
  expr.FieldRef(binding: 0, name: "title", lode_type: primitive.string())
}

Fetching records

Single record

Fetching record by ID

repo.get(repo: r, schema: movie_schema(), id: VInt(1))
// -> Ok(Some(Movie(..))) | Ok(None)

Fetching record by attributes

repo.get_by(repo: r, schema: movie_schema(), filters: [
  #("title", VString("Ready Player One")),
])

Fetching the first record

query.first(q, by: expr.col(field)) orders by the field ascending and takes one row (Ecto’s first/2) — or order and limit explicitly, then take one.

let q =
  query.from(source: "movies", alias: "m")
  |> query.order_by([query.asc(expr.col(movie_id()))])
  |> query.limit(1)
repo.one(repo: r, schema: movie_schema(), query: q)

Fetching the last record

Same, descending — query.last(q, by: expr.col(field)).

let q =
  query.from(source: "movies", alias: "m")
  |> query.order_by([query.desc(expr.col(movie_id()))])
  |> query.limit(1)
repo.one(repo: r, schema: movie_schema(), query: q)

Use `!` to raise if none is found

There are no bang variants — by design, misses are values (Ok(None)), not exceptions. When “absent is a bug”, assert deliberately:

let assert Ok(Some(movie)) =
  repo.get(repo: r, schema: movie_schema(), id: VInt(1))

Multiple records

Fetch all at once

repo.all(
  repo: r,
  schema: movie_schema(),
  query: query.from(source: "movies", alias: "m"),
)
// -> Ok([Movie(..), ..])

// Unfiltered? `repo.list` derives the query from the schema (no source string,
// no alias). Drop back to `repo.all` the moment you need a where/order_by/join.
repo.list(repo: r, schema: movie_schema())
// -> Ok([Movie(..), ..])

Stream all

repo.stream_fold walks a result set too large to materialize with all, folding rows in batches so only one batch is ever held in memory:

repo.stream_fold(
  repo: r,
  schema: movie_schema(),
  query: query.from(source: "movies", alias: "m"),
  batch_size: 500,
  from: 0,
  with: fn(count, _movie) { count + 1 },
)
// -> Ok(<total movies>), having fetched 500 rows at a time

repo.stream_for_each(repo:, schema:, query:, batch_size:, with:) is the side-effecting form. On Postgres this drives a real server-side cursor inside a transaction it opens for you (constant memory, one MVCC snapshot, batch_size rows per FETCH). The shape differs from Ecto — a fold rather than a lazy Enumerable, since Gleam has neither macros nor a lazy stream type — see #repo-stream.

Check at least one exists

repo.exists(repo: r, query: query.from(source: "movies", alias: "m"))
// -> Ok(True)

Relatedly, repo.count(repo: r, query: q) returns the matching row count.

Querying records

Keyword-based queries

There is no macro DSL, so Ecto’s keyword syntax (from m in Movie, where: ...) has no equivalent — every query is a value built with pipes (the “pipe-based” style below is the only style). Bindings are positional integers carried by the typed FieldRef accessors: the from source is binding 0, each join adds the next index.

let q =
  query.from(source: "movies", alias: "m")
  |> query.where(expr.eq(field: movie_title(), to: "Ready Player One"))
repo.all(repo: r, schema: movie_schema(), query: q)

Selecting specific columns is expressed with query.select:

query.from(source: "movies", alias: "m")
|> query.where(expr.eq(field: movie_title(), to: "Ready Player One"))
|> query.select([expr.col(movie_title()), expr.col(movie_tagline())])

Note: repo.all loads each row into the full schema struct, so a projection that omits required columns will fail to load — keep the default SELECT * when loading structs, and use projections with aggregates, col_as/ schema.load_prefixed joins, or the raw-SQL path below.

Interpolation with `^`

No pin operator needed: queries are plain data, so any Gleam variable is used directly. Every literal is rendered as a $n parameter — values are never spliced into the SQL text.

let title = "Ready Player One"
let q =
  query.from(source: "movies", alias: "m")
  |> query.where(expr.eq(field: movie_title(), to: title))
repo.all(repo: r, schema: movie_schema(), query: q)

Pipe-based queries

Already the native style — compose as many steps as you need:

let q =
  query.from(source: "movies", alias: "m")
  |> query.where(expr.eq(field: movie_title(), to: "Ready Player One"))
  |> query.order_by([query.desc(expr.col(movie_id()))])
  |> query.limit(10)
repo.all(repo: r, schema: movie_schema(), query: q)

Raw SQL escape hatch

For SQL the query model doesn’t cover (Ecto’s Repo.query/3), with positional $1, $2, ... parameters:

let assert Ok(rows) =
  repo.query_raw(
    repo: r,
    sql: "SELECT title, tagline FROM movies WHERE id = $1",
    params: [VInt(1)],
  )
// rows: List(Dict(String, Value)) — read columns by name,
// or load single fields with schema.load_field(row, "title", primitive.string())

To load full schema structs straight from a raw query in one call, use repo.query_raw_as(repo:, schema:, sql:, params:), which runs the SQL and loads each row through schema.load (columns matched by name).

Inserting records

Single record

Using internal data

repo.insert takes a changeset, so wrap the struct with changeset.change (no casting, no validations). Autogenerated columns (here id) are omitted from the INSERT and read back via RETURNING:

let cs =
  changeset.change(data: Person(id: 0, name: "Bob", age: 29), schema: person_schema())
let assert Ok(bob) = repo.insert(repo: r, schema: person_schema(), changeset: cs)
// bob.id is the database-assigned key

Using external data

Params are a Dict(String, Value) — your boundary (form/JSON decoding) produces the tagged Values:

// Params represent data from a form, API, CLI, etc.
let params = dict.from_list([#("name", VString("Bob")), #("age", VInt(29))])

let cs =
  changeset.cast(
    data: Person(id: 0, name: "", age: 0),
    schema: person_schema(),
    params: params,
    permitted: field.fields(["name", "age"]),
  )
  |> changeset.validate_required(field.fields(["name"]))
repo.insert(repo: r, schema: person_schema(), changeset: cs)
// an invalid changeset returns Error(ChangesetInvalid(errors: ..))

Multiple records

repo.insert_all takes typed rows (not maps), inserts them in one statement, and returns the count — no changesets, validations, or association writes:

let rows = [Person(0, "Bob", 29), Person(0, "Alice", 30)]
let assert Ok(2) = repo.insert_all(repo: r, schema: person_schema(), rows: rows)

Use repo.insert_all_returning(repo:, schema:, rows:) to get the stored rows back with generated keys filled in.

Upserts (`on_conflict`)

Where Ecto passes :on_conflict/:conflict_target options to Repo.insert, lode has a dedicated function taking a lode/on_conflict policy:

repo.upsert(
  repo: r,
  schema: person_schema(),
  changeset: cs,
  on_conflict: on_conflict.Update(
    target: on_conflict.Columns([field.field("name")]),     // given a unique index on name
    action: on_conflict.Replace([field.field("age")]),
  ),
)
// -> Ok(Some(stored_row)); with on_conflict.Nothing(target: ..),
//    a skipped duplicate comes back as Ok(None)

Columns are typed Field(row) values, not bare strings: field.field("name") names a column (the row is inferred from schema:, so a column of a different schema is a compile error). For typo-safety, use the codegen-emitted column record — person_fields.name — or expr.to_field(person_name()) to reuse a query accessor; either way the name is checked once, against the spec, not at every call site.

The bulk equivalents are repo.upsert_all and repo.upsert_all_returning.

Updating records

Single record

Using internal data

// fetch the row to update (Ecto's `first() |> Repo.one!()`)
let q = query.from(source: "people", alias: "p") |> query.limit(1)
let assert Ok(Some(person)) =
  repo.one(repo: r, schema: person_schema(), query: q)

let cs =
  changeset.change(data: person, schema: person_schema())
  |> changeset.put_change_typed(name: "age", of: primitive.integer(), value: 29)
let assert Ok(person) = repo.update(repo: r, schema: person_schema(), changeset: cs)

Using external data

// Params represent data from a form, API, CLI, etc.
let params = dict.from_list([#("age", VInt(29))])

let cs =
  changeset.cast(
    data: person,
    schema: person_schema(),
    params: params,
    permitted: field.fields(["age"]),
  )
repo.update(repo: r, schema: person_schema(), changeset: cs)

Insert or update

repo.insert_or_update picks insert or update for you — the find-or-create flow where you don’t know up front whether the row exists:

repo.insert_or_update(repo: r, schema: person_schema(), changeset: cs)
// inserts when the changeset's data has no primary key, updates when it does

Ecto reads the struct’s __meta__.state to decide; lode has no struct metadata, so it decides from primary-key presence instead — a meaningful divergence with a sharp edge for natural keys. See TASK:insert-or-update.

Multiple records (using queries)

repo.update_all(
  repo: r,
  query: query.from(source: "people", alias: "p"),
  set: [#("age", VInt(29))],
)
// -> Ok(affected_count)

Optimistic locking

changeset.optimistic_lock makes the write a compare-and-swap on a version column: the UPDATE/DELETE filters on the version read at load time, an update bumps it, and losing the race returns Error(error.StaleEntry).

let cs =
  changeset.change(person, person_schema())
  |> changeset.put_change_typed(field.field("age"), primitive.integer(), 30)
  |> changeset.optimistic_lock(field.field("lock_version"))
case repo.update(repo: r, schema: person_schema(), changeset: cs) {
  Ok(updated) -> ...                    // lock_version was bumped
  Error(error.StaleEntry) -> ...        // someone else wrote first: reload, retry
  Error(other) -> ...
}
// deletes carry the lock through the changeset-taking delete:
repo.delete_changeset(repo: r, schema: person_schema(), changeset:
  changeset.change(person, person_schema())
  |> changeset.optimistic_lock(field.field("lock_version")))

#optimistic-lock

Deleting records

Single record

let assert Ok(Some(person)) =
  repo.get(repo: r, schema: person_schema(), id: VInt(1))
let assert Ok(Nil) = repo.delete(repo: r, schema: person_schema(), row: person)

Multiple records (using queries)

repo.delete_all(repo: r, query: query.from(source: "people", alias: "p"))
// -> Ok(affected_count)

Multiple operations in one transaction

repo.transaction rolls back when the body returns Error. Use the transaction-scoped repo (tx) for every operation inside:

repo.transaction(repo: r, body: fn(tx) {
  use bob <- result.try(repo.insert(tx, person_schema(), bob_cs))
  use alice <- result.try(repo.insert(tx, person_schema(), alice_cs))
  Ok(#(bob, alice))
})

lode/multi composes named steps (Ecto’s Ecto.Multi); on failure the whole transaction rolls back and you get the failing step’s name:

let assert Ok(changes) =
  multi.new()
  |> multi.insert("bob", person_schema(), bob_cs)
  |> multi.insert("alice", person_schema(), alice_cs)
  |> multi.run("greet", fn(tx, changes) {
    let assert Ok(bob) = multi.get(changes, "bob")
    let bob: Person = bob          // multi results are Dynamic; annotate to recover
    send_welcome(tx, bob)
  })
  |> multi.transaction(r)

let assert Ok(bob) = multi.get(changes: changes, name: "bob")
let bob: Person = bob

multi covers Ecto’s core step set: insert/update/delete/run, bulk steps (multi.insert_all, multi.update_all, multi.delete_all — the stored result is the affected count), query steps (multi.one, multi.all, multi.exists), and multi.put. There is no dedicated insert_or_update/error/inspect step — a multi.run step covers each (call repo.insert_or_update on the transaction-scoped repo, return Error(..) to short-circuit, or print the results so far and pass them through). Compose multis with multi.merge (build steps from earlier results), multi.append, and multi.prepend. Step names must be unique: a duplicate makes multi.transaction return an error instead of running (Ecto raises at add/merge time — see #multi-breadth).

See also: Getting Started for the full setup and schema definitions, the Associations cheatsheet for put_assoc/ cast_assoc/repo.preload, and Divergences from Ecto for the complete list of intentional design differences.

Basic CRUD

Setup

Fetching records

Single record

Fetching record by ID

Fetching record by attributes

Fetching the first record

Fetching the last record

Use ! to raise if none is found

Multiple records

Fetch all at once

Stream all

Check at least one exists

Querying records

Keyword-based queries

Interpolation with ^

Pipe-based queries

Raw SQL escape hatch

Inserting records

Single record

Using internal data

Using external data

Multiple records

Upserts (on_conflict)

Updating records

Single record

Using internal data

Using external data

Insert or update

Multiple records (using queries)

Optimistic locking

Deleting records

Single record

Multiple records (using queries)

Multiple operations in one transaction

Use `!` to raise if none is found

Interpolation with `^`

Upserts (`on_conflict`)