DEV Community: Jakub Lambrych

How to report Postgres custom errors in Ecto Changeset

Jakub Lambrych — Thu, 04 Jul 2024 17:32:19 +0000

Sometimes you may find yourself in the need to capture a Postgres (or any other RDBMS) custom error in the Ecto.Changeset without raising an exception. This enables you to handle all the errors in one place without braking your aesthetic Elixir functional code with “try/rescue” constructs.

It has one big advantage: as the Ecto.Changeset is a “lingua franca” of many libraries and frameworks (like Phoenix), embedding error reports in the changeset struct will work for you out of the box without any additional error handing burden. Less code, less maintenance!

Under the hood

Currently, the Postgres Ecto Adapter (same as other adapters for major RDBMS) provide only limited support for reporting errors inside the Ecto Changeset. Let’s have a glimpse into the Postgres adapter source code:

@impl true
    def to_constraints(%Postgrex.Error{postgres: %{code: :unique_violation, constraint: constraint}}, _opts),
      do: [unique: constraint]
    def to_constraints(%Postgrex.Error{postgres: %{code: :foreign_key_violation, constraint: constraint}}, _opts),
      do: [foreign_key: constraint]
    def to_constraints(%Postgrex.Error{postgres: %{code: :exclusion_violation, constraint: constraint}}, _opts),
      do: [exclusion: constraint]
    def to_constraints(%Postgrex.Error{postgres: %{code: :check_violation, constraint: constraint}}, _opts),
      do: [check: constraint]

We can see that certain Postgres errors, namely those related to constraints, get a special treatment at the adapter level so that later could be transformed into relevant changeset errors on demand (by calling *_constraint functions in the changeset). Meanwhile, the remaining errors will be let through and propagated to your code. There are only few constraint error codes that get intercepted:

:unique_violation
:foreign_key_violation
:exclusion_violation
:check_violation

Solution

The method I would like to propose is to disguise your custom database error as one of the constraints that is already implemented by default in the Postgres Ecto adapter (see above).

In this example, I will define and raise a custom error from within a PL/pgSQL trigger function using Postgres’ check_contraint ERRCODE, but you can use any of the four, whichever makes more sense to you.

Step 1. Raise error in Postgres codebase.

CREATE FUNCTION custom_check() RETURNS TRIGGER AS $$
  BEGIN
    IF <SOME CONDITION> THEN
        RAISE EXCEPTION 'CUSTOM ERROR' 
        USING ERRCODE = 'check_violation', 
        CONSTRAINT = 'name_of_your_contraint';
    END IF;
    RETURN NEW;
  END;
  $$ LANGUAGE plpgsql

where:

CUSTOM ERROR is a custom string lateral of your choice that will be passed to Ecto as the error message text.
ERRCODE must be one of the following:
- unique_violation
- foreign_key_violation
- exclusion_violation
- check_violation
CONSTRAINT must have a name of your choice that will uniquely identify the custom error in the Ecto Changeset.

Please note: a comprehensive list of Postgres error codes can be found in the Postgres Documentation — Errors and Messages.

Step 2 Define standard constraint in Ecto Changeset.

In this case, I consistently follow check_contraint error code raised in Postgres and call check_constraint function in the changeset to capture it.

def changeset(schema, attrs) do
   schema
   |> check_constraint(:some_field, name: name_of_your_contraint: , message: "custom error message")
end

where:

:some_field is a key of associated with the model struct. It is particularly useful when working with Phoenix forms.
:name_of_your_contraint is an atom reflecting the same name as the one defined in the Postgres codebase.
message is an error message on Ecto side that will from additional contextual information.

To make your Elixir code more readable, you could consider some refactoring:

def changeset(schema, attrs) do
   schema
   |> my_custom_error(:some_field)
end

defp my_custom_error(schema, key) do
  schema
  |> check_constraint(key, name: name_of_your_contraint: , message: "custom error message")
 end

A minor trade off is that a potential error description in the changeset has to be related to a key in the existing Schema struct. This is because changesets are designed on field level. If you use Phoenix form, you can compensate this drawback with an accurate error message propagated to the user.

Summary

In this article, I tried to propose a fairly easy technique to intercept custom database errors and turn them into a Ecto Changeset errors. This all without the need to override your Repo module functionality nor forking the adapter’s code, which would be way more difficult to maintain with new Ecto library updates.

Please, feel free to leave your comment and share other approaches that you came across.

Elixir: map casting to structs (with checks at compile time!)

Jakub Lambrych — Mon, 10 Jun 2024 08:04:28 +0000

Intro

In this article you will learn:

how to leverage compile-time validation of maps to be merged later with %struct{}, that can save you time and hassle after the deployment
review popular runtime validation functions
how to use a macro to do the same at compile time (!)
import KeyValidator library to your project from HEX: https://hex.pm/packages/key_validator

Runtime validations

Elixir and Ecto has built-in functions that perform the key validity check of maps, but only at runtime:

Kernel.struct/2
Kernel.struct!/2
Ecto.Query.select_merge/3

Kernel.struct!/2

Let's take a look at the following example:

defmodule User do
defstruct name: "john"
end

# Following line is a runtime only check:

Kernel.struct!(User, %{name: "Jakub"})
#=> %User{name: "Jakub"}

# Runtime error on key typo:

Kernel.struct!(User, %{nam__e: "Jakub"})
#=> ** (KeyError) key :nam__e not found

The expression Kernel.struct!(User, %{name: "Jakub"}) uses a map literal (%{name: "Jakub"}). The User struct definition is known beforehand, as well as the map structure. However, the comparison between the keys in the User struct and the map literal will only take place during when the app is running and the code getting actually executed. Thus, any potential typo in the key will be discovered only at runtime.

Ecto.Query.API.select_merge/3

A similar situation takes place when using a popular Ecto.Query.select_merge/3 for populating virtual fields in schemas:

defmodule Post do
    use Ecto.Schema
    schema "posts" do
      field :author_firstname, :string
      field :author_lastname, :string
      field :author, :string, virtual_field: true
    end
end

defmodule Posts do
    import KeyValidator

    def list_posts do
      Post
      |> select_merge([p], %{author: p.author_firstname <> " " <> p.author_lastname})
      |> Repo.all()
    end
end

Posts.list_posts()

In the provided example, Post schema contains a :author virtual field. We want to store a concatenated value of the post's author first name and the last name populated in the query. The select_merge will merge the given expression with query results.

If the result of the query is a struct (in our case Post) and in we are merging a map (in our case %{author: data} literal), we need to assert that all the keys from the map exist in the struct. To achieve this, Ecto uses Ecto.Query.API.merge/2 underneath:

If the map on the left side is a struct, Ecto will check all the fields on the right previously exist on the left before merging.

This, however, is done in runtime only again. You will learn whether the map keys conform to the struct key when the particular line of code got executed.

Compile-time validation

In certain situations, the conformity between map/keyword keys could be already checked at the compile-time.

One example when we can potentially leverage the compile-time validations is when we work with map/keyword literals in our code. These literals are provided directly in the Elixir's AST structure during compilation process. We can build on this fact if our intention is to use the map for casting onto structs at some point in our code.

We deal with map or keyword literals when the structure is given inline:

# Map literal:

%{name: "Jakub", lastname: "Lambrych"}

# Keyword literal

[name: "Jakub", lastname: "Lambrych"]

In contrast, the following expressions do not allow us to work with literals when invoking merging maps with struct:


# map is assigned to a variable
m = %{name: "Artur"}

# Function invoked with variable "m" and not a "map literal".
# No opportunity for compile check.
Kernel.struct(User, m)

Use `KeyValidator` macro for compile-time checks

In the following example, User module together with the map literal is defined at the compile time. We will leverage the power of compile-time macros whenever our intention is to use a particular map/keyword literal to be merged at some point with a struct.

To support this, I developed a small KeyValidator.for_struct/2 macro (available on Hex). Let's see how we can use it:

defmodule User do
defstruct name: "john"
end

import KeyValidator

# Succesfull validation. Returns the map:

user_map = for_struct(User, %{name: "Jakub"})
#=> %{name: "Jakub"}

Kernel.struct!(User, user_map)
#=> %User{name: "Jakub"}

# Compile time error on "nam__e:" key typo:

user_map2 = for_struct(User, %{nam__e: "Jakub"})
#=>** (KeyError) Key :name_e not found in User

The KeyValidator.for_struct/2 can be also used with Ecto.Query.select_merge/3 to populate virtual_fields:

defmodule Posts do
    import KeyValidator

    def list_posts do
        Post
        |> select_merge([p], for_struct(Post, %{authorrr: "Typo"}))
        |> Repo.all()
      end
end


Posts.list_posts()
#=> ** (KeyError) Key :authorrr not found in Post

The above code will raise a Key Error during compilation.

Install `KeyValidator` from Hex

For convenience, I wrapped the macro in library. It is available on hex with documentation:

https://hex.pm/packages/key_validator

You can add it as a dependency to your project:

def deps do
  [
    {:key_validator, "~> 0.1.0", runtime: false}
  ]
end

Source code

Macro code together with ExUnit tests can be accessed at the GitHub repo:

https://github.com/utopos/key_validator

Summary

Using metaprogramming in Elixir (macros) give you an opportunity to analyze the Elixir's AST and perform some checks of structs, maps and keywords before deploying your code.

Adding KeyValidator.for_struct/2 macro to your project, allows some category of key conformity errors to be caught at an early stage in the development workflow. No need to wait until the code to crashes at runtime. It can be expensive and time-consuming to fix bugs caused by simple typos - i.e. when map keys are misspelled.

Although the macro usage covers some scenarios, we need to bear in mind that it is not a silver bullet. The KeyValidator cannot accept dynamic variables due to the nature of Elixir macros. Only map/keyword literals are accepted as their content can be accessed during AST expansion phase of Elixir compilation process.

Adding stream_async() to Phoenix LiveView

Jakub Lambrych — Thu, 06 Jun 2024 16:42:15 +0000

In this article, you will learn:

how to implement in LiveView asynchronous assign for streams
how to use the async_result function component (UI) for streams in Heex templates
how to use meta programming to auto generate boilerplate for asynchronous stream handling - stream_async() macro
simply add stream_async via Hex package

New asynchronous operations in LiveView

A new release of LiveView library — v 0.20.0 — introduced built-in functions for asynchronous work. It's a perfect solution to deliver a snappy user experience by delegating some time-consuming tasks (ex. fetching from external services) to background jobs without blocking the UI or event handlers.

assign_sync/3 - a straight forward way to load the results asynchronously from these background tasks into socket assigns.
start_async/4 - a more granular control over async task result handling.
<.async_result ...> - component to handle the asynchronous operation state on the UI side - Heex templates (for success, loading, and errors).

Missing companion: `stream_async()`

Sometimes you may need to work asynchronously with streams.

Streaming results allow working with large collections without keeping them on the server. This functionality is not available out of the box yet.

I will demonstrate further on how to manually implement an asynchronous stream assign to the LivewView socket and wrap the whole boilerplate into a reusable stream_async() macro.

Asynchronous streaming in LiveView

In the following example, LiveView loads a list of hotels in a specified location. To load the data, we will use Hotels.fetch(location) function. Here's the code:

@hotels :hotels

def mount(%{"location" => location}, _, socket) do
  {:ok,
   socket
   |> assign(@hotels, AsyncResult.loading())
   |> start_async(@hotels, fn -> Hotels.fetch!(location) end)
   }
end

def handle_async(@hotels, {:ok, hotels}, socket) do
   {:noreply,
    socket
    |> assign(@hotels, AsyncResult.ok(@hotels))
    |> stream(@hotels, hotels, reset: true)
    }
end

def handle_async(@hotels, {:exit, reason}, socket) do
   {:noreply, 
    update(@hotels, fn async_result -> AsyncResult.failed(async_result, {:exit, reason}) end)
    }
end

The core of the solution in the presented code revolves around using a pair of assigns:

socket.assigns.hotels: instance of %AsyncResult{ } struct handling the async result.
- (line: assign(@hotels, AsyncResult.loading()))
socket.assigns.streams.hotels - stream for target large collection
- (line: stream(@hotels, hotels, reset: true)).

I use two different maps in the socket to store necessary data (assigns and the nested map: assignes.streams). Note I use the same key in both: :hotels (the reason will become clear in a moment). Using a combination of a direct assign and a stream assign solves the following challenge: how to store async loading state for a stream that is not yet populated.

According to the documentation, a stream assign must contain only collections and nothing else. We cannot store temporarily any other type of data (here: loading state) as it will produce errors. Thus, need an additional assign to inform whether the stream content is ready to render, or maybe there was an error and stream is not assigned.

It might look tempting to resign from using %AsyncResult{} assign and simply stream en empty collection, meanwhile we are fetching "the real data" asynchronously. The role of the empty collection would be to signal "loading state" for the duration of the async operation. Personally, I don't find this approach right. An empty stream may indicate to the UI and the user that there's no data returned as a response to their request (in the example: there are no hotels in the indicated location). Furthermore, we would lose the ability to differentiate between "loading state" and "failed state" of collection fetching. As a result, we lose as well the opportunity to provide a meaningful error message to the user.

Anatomy of the solution

Let's dig deeper into the proposed code:

start_async/4 function - asynchronous task wrapper. Used to get result that will be later streamed.
handle_async - 2x callbacks on the LiveView process to deal with start_async/4 task results:
- {:ok, results} - success; collection to stream available in results variable,
- {:exit, reeason}- failure.
Phoenix.LiveView.AsyncResult - LiveView struct to track state of an async assign.
- Set the state and assign results on the struct via functions: ok(), loading() and failed()
- Read the state by accessing three boolean state fields : :ok? (success), :loading and :failed.
- Read the results of the async operation be accessing :result field.

Accessing async stream in Heex

When the async collection loading is successful, our socket structure will look like this:

%{
  Stream: #Phoenix.LiveView.Socket<
  id: "phx-F9YIUoGp9R8towwB",
  [...]
  assigns: %{
      hotels: %Phoenix.LiveView.AsyncResult{
      ok?: true,
      loading: nil,
      failed: nil,
      result: :hotels
    },
    streams: %{
      hotels: %Phoenix.LiveView.LiveStream{
        name: :hotels,
        [...]

Pay attention that use the same atom (:hotels) as:

key to access socket.assigns referring to the async state data structure (%AsyncResult{}),
result value of the AsyncResult.result field,
key to access socket.assigns.streams where the stream for the large collection eventually ends up.

Although it may look confusing at a first glance, you will see soon that this helps us to produce a highly reusable code in LiveView'srender() to access the stream:

def render(assigns) do
~H"""
<.async_result :let={stream_key} assign={@hotels}>
  <:loading>Loading hotels...</:loading>
  <:failed :let={_failure}>There was an error loading the hotels. Please try again later.</:failed>
      <ul id="hotels_stream" phx-update="stream">
        <li :for={{id, hotel} <- @streams[stream_key]} id={id}>
          <%= hotel.name %>
        </li>
      </ul>
</.async_result>
"""
end

LiveView built-in <.async_result ...> component is designed to work with the %AsyncResult{} structs.
%AsyncStruct{} is passed via "assign={}" attribute of the component.
<.async_result ...> component @inner_block receives stream_key which is used to fetch the correct stream from the socket.assigns.streams (accessed by @streams in the code). The stream_key becomes available in the @inner_block via :let={} attribute.

Use `stream_async()` function

As we could observe, working manually with steams asynchronously adds a repetitive boilerplate to our LiveView. For every collection that we would like to stream, we need to add explicitly handle_async() callbacks.

The solution proposed in this article can be easily wrapped in a macro that will auto-generate all the necessary handling callbacks behind the scenes.

The stream_async/4 macro can be used as follows:

  use LiveStreamAsync

  def mount(%{"location" => location}, _, socket) do
    {:ok,
     socket
     |> stream_async(:hotels, fn -> Hotels.fetch!(location) end)
    }
  end

The stream_async/4 macro supports opts keyword. Options will be piped accordingly to the start_async/4 and stream/3 functions. To learn more about these options, check the official LiveView documentation (in the reference section at the end of the article).

We're rendering results in Heex template same as before (repeating):

def render(assigns) do
~H"""
<.async_result :let={stream_key} assign={@hotels}>
  <:loading>Loading hotels...</:loading>
  <:failed :let={_failure}>There was an error loading the hotels. Please try again later.</:failed>
      <ul id="hotels_stream" phx-update="stream">
        <li :for={{id, hotel} <- @streams[stream_key]} id={id}>
          <%= hotel.name %>
        </li>
      </ul>
</.async_result>
"""
end

Macro code

Just add the following macro to your project:

defmodule LiveStreamAsync do
  alias LiveView.AsyncResult

  defmacro __using__(_opts) do
    quote do
      import unquote(__MODULE__)
      @before_compile unquote(__MODULE__)
      Module.register_attribute(__MODULE__, :async_streams, accumulate: true)
    end
  end

  defmacro __before_compile__(_env) do
    streams = Module.get_attribute(__CALLER__.module, :async_streams)

    for {stream_id, opts} <- streams do
      quote bind_quoted: [stream: stream_id, opts: opts] do
        def handle_async(stream, {:ok, results}, socket) do
          socket =
            socket
            |> assign(stream, AsyncResult.ok(stream))
            |> stream(stream, results, unquote(opts))

          {:noreply, socket}
        end

        def handle_async(stream, {:exit, reason}, socket) do
          {:noreply,
           update(socket, stream, fn async_result ->
             AsyncResult.failed(async_result, {:exit, reason})
           end)}
        end
      end
    end
  end

  defmacro stream_async(socket, key, func, opts \\ []) do
    Module.put_attribute(__CALLER__.module, :async_streams, {key, opts})

    quote bind_quoted: [socket: socket, key: key, func: func, opts: opts] do
      socket
      |> assign(key, AsyncResult.loading())
      |> start_async(key, func, opts)
    end
  end
end

Hex package `live_stream_async`

For ease, I packaged this macro on hex.pm, so you can easily add as your project dependency in the mix.exs:

defp deps do
    [
      {:live_stream_async, "~> 0.1.0", runtime: false}
    ]

Use it the same way as described in this article Use stream_async() function

Enjoy!

Summary

In this article, we learned:

new functions in LiveView v 0.20.0 to work with asynchronous tasks
fetching big collection for streaming require using low level start_async/4 function combined with handle_async() callbacks
combining %AsyncResult{} struct with async streaming allows to control the state of loading big collections in the UI
import stream_async() functionality to your project with hex.pm.

DEV Community: Jakub Lambrych

How to report Postgres custom errors in Ecto Changeset

Under the hood

Solution

Step 1. Raise error in Postgres codebase.

Step 2 Define standard constraint in Ecto Changeset.

Summary

Elixir: map casting to structs (with checks at compile time!)

Intro

Runtime validations

Kernel.struct!/2

Ecto.Query.API.select_merge/3

Compile-time validation

Use KeyValidator macro for compile-time checks

Install KeyValidator from Hex

Source code

Summary

Adding stream_async() to Phoenix LiveView

New asynchronous operations in LiveView

Missing companion: stream_async()

Asynchronous streaming in LiveView

Anatomy of the solution

Accessing async stream in Heex

Use stream_async() function

Macro code

Hex package live_stream_async

Summary

References

Use `KeyValidator` macro for compile-time checks

Install `KeyValidator` from Hex

Missing companion: `stream_async()`

Use `stream_async()` function

Hex package `live_stream_async`