OpenAI Image Generation in Elixir: Build an AI Image Generator in Phoenix LiveView

Text generation gets all the tutorials. Image generation is the feature customers actually screenshot and share, and it is surprisingly easy to get wrong in a web app: a single request can take thirty seconds or more, the response is a few megabytes of base64, and every image costs real money. If you call the API from a LiveView handle_event/3 and hope for the best, you will block the process, lose the result on a refresh, and have no idea what each customer spent.

This guide does OpenAI image generation in Elixir properly. We will call the Images API with Req, store the result, run the slow part in an Oban job, push progress to the browser over PubSub, and cap how many images each user can make. It is part of our series on building AI apps with Elixir and Phoenix, and everything here works on Phoenix 1.7 and 1.8.

What we are building

A LiveView page with a prompt box, a size and quality picker, and a grid of the user’s images. When the user submits a prompt:

1. A generations row is inserted with status pending, and a placeholder card appears instantly.
2. An Oban job calls the OpenAI Images API, decodes the base64 image and writes it to storage.
3. The job marks the row done (or failed with a reason) and broadcasts on PubSub.
4. The LiveView receives the broadcast and swaps the placeholder for the finished image.

The user can close the tab and come back later. The image is still there, because nothing about the work depends on the LiveView process staying alive.

The Images API in one request

The endpoint is POST https://api.openai.com/v1/images/generations. The request body is small:

{
  "model": "gpt-image-1",
  "prompt": "A watercolor fox reading a newspaper in a Paris cafe",
  "size": "1024x1024",
  "quality": "medium",
  "n": 1
}

With the GPT image models the response carries the image itself, base64 encoded, rather than a URL:

{
  "created": 1759740000,
  "data": [{ "b64_json": "iVBORw0KGgoAAAANSUhEUgAA..." }],
  "usage": { "input_tokens": 18, "output_tokens": 1056, "total_tokens": 1074 }
}

A few things to know before writing any code:

• Model names move. OpenAI ships new image models regularly. Keep the model in config so you can switch without a deploy, and check the models page for the current one.
• Sizes are fixed. The GPT image models accept 1024x1024, 1536x1024 (landscape), 1024x1536 (portrait) and auto. Anything else is a 400.
• Quality drives cost and latency. low, medium and high differ a lot in both. Default to medium and let paying users pick high.
• Older DALL-E models return URLs that expire after about an hour. If you use one, download the file immediately. Never store an OpenAI URL in your database.
• Prompts are moderated. A rejected prompt comes back as a 400 with an error body. That is a normal outcome you have to show the user, not an exception.

Setup

Add Req and Oban if you do not have them already:

# mix.exs
defp deps do
  [
    {:req, "~> 0.5"},
    {:oban, "~> 2.18"},
    # ...
  ]
end

Put the key and the defaults in runtime config, never in source:

# config/runtime.exs
config :my_app, :openai,
  api_key: System.fetch_env!("OPENAI_API_KEY"),
  image_model: System.get_env("OPENAI_IMAGE_MODEL", "gpt-image-1")

A small Images client

Keep the HTTP call in one module. It returns plain tuples so the caller never has to know about Req.

defmodule MyApp.AI.Images do
  @moduledoc "Thin client for the OpenAI Images API."

  @sizes ~w(1024x1024 1536x1024 1024x1536)
  @qualities ~w(low medium high)

  def sizes, do: @sizes
  def qualities, do: @qualities

  @spec generate(String.t(), keyword()) ::
          {:ok, %{binary: binary(), usage: map()}} | {:error, term()}
  def generate(prompt, opts \\ []) do
    config = Application.fetch_env!(:my_app, :openai)

    body = %{
      model: Keyword.get(opts, :model, config[:image_model]),
      prompt: prompt,
      size: Keyword.get(opts, :size, "1024x1024"),
      quality: Keyword.get(opts, :quality, "medium"),
      n: 1
    }

    req =
      Req.new(
        base_url: "https://api.openai.com/v1",
        auth: {:bearer, config[:api_key]},
        # image generation is slow: give it room
        receive_timeout: 120_000,
        retry: :transient
      )
      |> Req.merge(Application.get_env(:my_app, :images_req_options, []))

    case Req.post(req, url: "/images/generations", json: body) do
      {:ok, %{status: 200, body: %{"data" => [%{"b64_json" => b64} | _]} = resp}} ->
        {:ok, %{binary: Base.decode64!(b64), usage: resp["usage"] || %{}}}

      {:ok, %{status: 400, body: %{"error" => %{"message" => message} = error}}} ->
        {:error, {:rejected, error["code"], message}}

      {:ok, %{status: 429}} ->
        {:error, :rate_limited}

      {:ok, %{status: status, body: body}} ->
        {:error, {:http, status, body}}

      {:error, exception} ->
        {:error, exception}
    end
  end
end

Three details matter here. receive_timeout: 120_000 overrides Req’s default, which is far too short for a high quality image. retry: :transient retries network errors and 5xx responses but not a 400, so a rejected prompt is not billed twice. And the :images_req_options merge is a hook for tests, which we will use at the end.

If you are already handling 429s across several AI calls, the patterns in OpenAI rate limits in Elixir apply unchanged: image endpoints have their own, lower limits.

The schema

One table records every request, whether it succeeded or not. That row is your audit trail, your history page and your cost report.

defmodule MyApp.Repo.Migrations.CreateGenerations do
  use Ecto.Migration

  def change do
    create table(:generations) do
      add :user_id, references(:users, on_delete: :delete_all), null: false
      add :prompt, :text, null: false
      add :size, :string, null: false
      add :quality, :string, null: false
      add :status, :string, null: false, default: "pending"
      add :path, :string
      add :error, :text
      add :usage, :map, default: %{}
      timestamps()
    end

    create index(:generations, [:user_id, :inserted_at])
  end
end

defmodule MyApp.AI.Generation do
  use Ecto.Schema
  import Ecto.Changeset

  schema "generations" do
    belongs_to :user, MyApp.Accounts.User
    field :prompt, :string
    field :size, :string, default: "1024x1024"
    field :quality, :string, default: "medium"
    field :status, Ecto.Enum, values: [:pending, :done, :failed], default: :pending
    field :path, :string
    field :error, :string
    field :usage, :map, default: %{}
    timestamps()
  end

  def create_changeset(generation, attrs) do
    generation
    |> cast(attrs, [:prompt, :size, :quality])
    |> validate_required([:prompt])
    |> validate_length(:prompt, min: 3, max: 1_000)
    |> validate_inclusion(:size, MyApp.AI.Images.sizes())
    |> validate_inclusion(:quality, MyApp.AI.Images.qualities())
  end
end

Storing the image

The API hands you bytes, so you choose where they live. For a real app that is object storage. The function below writes to S3 through ExAws, and the comment shows the local disk version for development.

defmodule MyApp.AI.ImageStore do
  @bucket "my-app-generations"

  @spec put(integer(), binary()) :: {:ok, String.t()} | {:error, term()}
  def put(generation_id, binary) do
    key = "generations/#{generation_id}-#{System.unique_integer([:positive])}.png"

    @bucket
    |> ExAws.S3.put_object(key, binary, content_type: "image/png")
    |> ExAws.request()
    |> case do
      {:ok, _} -> {:ok, key}
      {:error, reason} -> {:error, reason}
    end

    # In development, write under priv/static instead:
    # path = Path.join([:code.priv_dir(:my_app), "static", "generated", key])
    # File.mkdir_p!(Path.dirname(path))
    # File.write!(path, binary)
  end

  def url(key), do: "https://#{@bucket}.s3.amazonaws.com/#{key}"
end

If you serve these to other users, use presigned URLs or a CDN in front of a private bucket instead of a public one. The direct-to-S3 upload guide covers the bucket policy side.

The context: create the row, enqueue the job

defmodule MyApp.AI do
  import Ecto.Query
  alias MyApp.Repo
  alias MyApp.AI.{Generation, GenerateImageWorker}

  @daily_limit 20

  def topic(user_id), do: "generations:#{user_id}"

  def list_generations(user) do
    Repo.all(
      from g in Generation,
        where: g.user_id == ^user.id,
        order_by: [desc: g.inserted_at],
        limit: 60
    )
  end

  def change_generation(attrs \\ %{}),
    do: Generation.create_changeset(%Generation{}, attrs)

  def request_image(user, attrs) do
    if used_today(user) >= @daily_limit do
      {:error, :daily_limit}
    else
      Ecto.Multi.new()
      |> Ecto.Multi.insert(:generation, fn _ ->
        %Generation{user_id: user.id} |> Generation.create_changeset(attrs)
      end)
      |> Oban.insert(:job, fn %{generation: g} ->
        GenerateImageWorker.new(%{generation_id: g.id})
      end)
      |> Repo.transaction()
      |> case do
        {:ok, %{generation: generation}} -> {:ok, generation}
        {:error, :generation, changeset, _} -> {:error, changeset}
        {:error, _step, reason, _} -> {:error, reason}
      end
    end
  end

  defp used_today(user) do
    since = DateTime.utc_now() |> DateTime.add(-1, :day)

    Repo.aggregate(
      from(g in Generation,
        where: g.user_id == ^user.id and g.inserted_at > ^since and g.status != :failed
      ),
      :count
    )
  end
end

The Ecto.Multi matters: the row and the job commit together or not at all, so you never have a pending row with no job behind it. If Oban is new to you, the Oban background jobs guide explains queues, retries and uniqueness.

The @daily_limit count query is fine for a side project. It is not fine once images are a paid feature: it reads Postgres on every request, two quick submits can both pass the check, and it says nothing about billing. That is the problem Aurora Meter exists for. It keeps the counter in ETS, and AuroraMeter.with_quota(user, :images, fn -> ... end) reserves a unit atomically before the work runs, so two concurrent requests cannot both take the last image of a plan.

The worker

defmodule MyApp.AI.GenerateImageWorker do
  use Oban.Worker, queue: :images, max_attempts: 3

  alias MyApp.{Repo, AI}
  alias MyApp.AI.{Generation, Images, ImageStore}

  @impl Oban.Worker
  def timeout(_job), do: :timer.minutes(3)

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"generation_id" => id}, attempt: attempt, max_attempts: max}) do
    generation = Repo.get!(Generation, id)

    with {:ok, %{binary: png, usage: usage}} <-
           Images.generate(generation.prompt, size: generation.size, quality: generation.quality),
         {:ok, key} <- ImageStore.put(generation.id, png) do
      generation
      |> Ecto.Changeset.change(status: :done, path: key, usage: usage)
      |> Repo.update!()
      |> broadcast()

      :ok
    else
      # the prompt was refused: retrying will not help
      {:error, {:rejected, _code, message}} ->
        fail(generation, message)
        {:cancel, :rejected}

      {:error, reason} when attempt >= max ->
        fail(generation, "Generation failed, please try again.")
        {:error, reason}

      {:error, reason} ->
        {:error, reason}
    end
  end

  defp fail(generation, message) do
    generation
    |> Ecto.Changeset.change(status: :failed, error: message)
    |> Repo.update!()
    |> broadcast()
  end

  defp broadcast(generation) do
    Phoenix.PubSub.broadcast(MyApp.PubSub, AI.topic(generation.user_id), {:generation, generation})
  end
end

Give images their own queue with a low concurrency limit, for example queues: [default: 10, images: 3]. That keeps a burst of image jobs from starving your other work and keeps you under OpenAI’s per-minute image limit without any extra code.

Returning {:cancel, :rejected} for a moderation refusal is deliberate. A refused prompt will be refused again, and each retry would cost an API call.

The LiveView

defmodule MyAppWeb.ImageLive do
  use MyAppWeb, :live_view

  alias MyApp.AI
  alias MyApp.AI.{Images, ImageStore}

  def mount(_params, _session, socket) do
    user = socket.assigns.current_user
    if connected?(socket), do: Phoenix.PubSub.subscribe(MyApp.PubSub, AI.topic(user.id))

    {:ok,
     socket
     |> assign(form: to_form(AI.change_generation()))
     |> stream(:generations, AI.list_generations(user))}
  end

  def handle_event("generate", %{"generation" => params}, socket) do
    case AI.request_image(socket.assigns.current_user, params) do
      {:ok, generation} ->
        {:noreply,
         socket
         |> stream_insert(:generations, generation, at: 0)
         |> assign(form: to_form(AI.change_generation()))}

      {:error, :daily_limit} ->
        {:noreply, put_flash(socket, :error, "You have used today's images. Upgrade for more.")}

      {:error, %Ecto.Changeset{} = changeset} ->
        {:noreply, assign(socket, form: to_form(changeset))}

      {:error, _} ->
        {:noreply, put_flash(socket, :error, "Could not start the generation.")}
    end
  end

  def handle_info({:generation, generation}, socket) do
    {:noreply, stream_insert(socket, :generations, generation)}
  end
end

stream_insert/3 with an existing id replaces the card in place, so the same call handles both the new placeholder and the finished image. The template:

<.simple_form for={@form} phx-submit="generate">
  <.input field={@form[:prompt]} type="textarea" label="Describe the image" />
  <.input field={@form[:size]} type="select" options={Images.sizes()} label="Size" />
  <.input field={@form[:quality]} type="select" options={Images.qualities()} label="Quality" />
  <:actions><.button phx-disable-with="Queuing...">Generate</.button></:actions>
</.simple_form>

<div id="generations" phx-update="stream" class="grid grid-cols-2 md:grid-cols-3 gap-4 mt-8">
  <div :for={{id, g} <- @streams.generations} id={id} class="rounded-lg border p-2">
    <%= case g.status do %>
      <% :pending -> %>
        <div class="aspect-square animate-pulse bg-zinc-100 rounded"></div>
      <% :done -> %>
        <img src={ImageStore.url(g.path)} alt={g.prompt} class="rounded" loading="lazy" />
      <% :failed -> %>
        <p class="text-sm text-red-600"><%= g.error %></p>
    <% end %>
    <p class="mt-2 text-xs text-zinc-500 line-clamp-2"><%= g.prompt %></p>
  </div>
</div>

The alt attribute uses the prompt, which is free accessibility for every image.

When Oban is more than you need

For an internal tool where losing an in-flight image on a deploy is acceptable, LiveView’s start_async/3 (LiveView 0.20 and later) runs the call in a task linked to the socket and is less code:

def handle_event("generate", %{"prompt" => prompt}, socket) do
  {:noreply,
   socket
   |> assign(:loading, true)
   |> start_async(:image, fn -> Images.generate(prompt) end)}
end

def handle_async(:image, {:ok, {:ok, %{binary: png}}}, socket) do
  {:noreply, assign(socket, loading: false, data_url: "data:image/png;base64," <> Base.encode64(png))}
end

def handle_async(:image, {:ok, {:error, reason}}, socket) do
  {:noreply, socket |> assign(loading: false) |> put_flash(:error, inspect(reason))}
end

The tradeoff is honest: the task dies with the LiveView. Close the tab and you paid for an image nobody will see. For anything customer facing, the job version wins.

Costs, limits and abuse

Image generation is one of the more expensive API calls you can make per click, so plan for it on day one:

• Cap per user per period. The daily limit above is the minimum. Tie it to the plan: free users get a handful, paid users get more, high quality counts for more than low.
• Record usage. The usage map from the response is stored on every row, so you can work out exactly what each user cost you last month with one SUM query.
• Limit concurrency. The Oban queue limit is your global brake. One user should not be able to hold all three slots, so add a per-user check for pending rows.
• Keep prompts. Storing every prompt lets you answer abuse reports and tune moderation. Mention it in your privacy policy.

Testing without spending money

Req ships a test adapter, so the client can be exercised without touching the network. Point the client at a stub in the test config:

# config/test.exs
config :my_app, :images_req_options, plug: {Req.Test, MyApp.AI.Images}
config :my_app, :openai, api_key: "test", image_model: "gpt-image-1"

defmodule MyApp.AI.ImagesTest do
  use ExUnit.Case, async: true
  alias MyApp.AI.Images

  @png <<137, 80, 78, 71, 13, 10, 26, 10>>

  test "decodes the base64 image" do
    Req.Test.stub(Images, fn conn ->
      Req.Test.json(conn, %{"data" => [%{"b64_json" => Base.encode64(@png)}], "usage" => %{}})
    end)

    assert {:ok, %{binary: @png}} = Images.generate("a fox")
  end

  test "a refused prompt is an error, not a crash" do
    Req.Test.stub(Images, fn conn ->
      conn
      |> Plug.Conn.put_status(400)
      |> Req.Test.json(%{"error" => %{"code" => "moderation_blocked", "message" => "Refused"}})
    end)

    assert {:error, {:rejected, "moderation_blocked", "Refused"}} = Images.generate("bad")
  end
end

For the worker, Oban.Testing with testing: :manual lets you assert the job was enqueued, then perform_job/2 runs it against the same stub.

Summary

OpenAI image generation in Elixir comes down to four decisions:

1. Treat the call as slow and expensive. A long receive_timeout, retries only on transient errors, and no retries on a refused prompt.
2. Persist before you call. A pending row plus an Oban job in one transaction means a refresh or a deploy never loses an image the user paid for.
3. Push, do not poll. PubSub plus stream_insert/3 updates the grid the moment the job finishes.
4. Meter it. Store the usage on every row and cap images per plan, because this is the feature that will run up your bill.

If you would rather start from a working AI app than wire the API client, job queue and LiveView streaming yourself, phx_ai ships an OpenAI-backed Phoenix app with auth, streaming responses and a clean place to add an image generator like this one. And if you plan to build more than one product, the Builder Pass gives you every PhxTemplates starter, current and future, for a single lifetime price.