Kathikon.Batch (Kathikon v0.2.0)

Copy Markdown View Source

Simple parent/child batch workflows for fan-out/fan-in.

The parent job moves to :waiting_for_children without blocking a BEAM process. When the batch completes, an explicit continuation job is enqueued.

Examples

{:ok, parent} = Kathikon.Storage.insert(parent_job)

{:ok, batch} =
  Kathikon.Batch.start(parent.id, [
    {ChildWorker, %{"id" => 1}, [queue: :default]},
    {ChildWorker, %{"id" => 2}, [queue: :default]}
  ], on_complete: {ReportWorker, %{"parent_id" => parent.id}})

{:ok, %{status: :running}} = Kathikon.Batch.status(batch.batch_id)

See docs/batches.md.

Summary

Functions

Lists child job ids for a batch.

Returns results for completed child jobs in a batch.

Retries failed children in a batch.

Starts a batch from a parent job, enqueueing child jobs.

Returns batch status by batch id.

Types

child_spec()

@type child_spec() :: {module(), term(), keyword()} | map()

Functions

children(batch_id)

@spec children(String.t()) :: {:ok, [String.t()]} | {:error, term()}

Lists child job ids for a batch.

Examples

{:ok, child_ids} = Kathikon.Batch.children(batch_id)

results(batch_id)

@spec results(String.t()) :: {:ok, [map()]} | {:error, term()}

Returns results for completed child jobs in a batch.

Examples

{:ok, results} = Kathikon.Batch.results(batch_id)

Enum.filter(results, &(&1.state == :completed))

retry_failed(batch_id)

@spec retry_failed(String.t()) :: {:ok, [Kathikon.Job.t()]} | {:error, term()}

Retries failed children in a batch.

Examples

{:ok, retried} = Kathikon.Batch.retry_failed(batch_id)
length(retried)

start(parent_job_id, child_specs, opts \\ [])

@spec start(String.t(), [child_spec()], keyword()) :: {:ok, map()} | {:error, term()}

Starts a batch from a parent job, enqueueing child jobs.

Options

  • :on_complete{WorkerModule, args} continuation when batch succeeds
  • :success_policy:all_succeeded (default), {:at_least, n}, or :allow_partial
  • :queue — queue for child jobs

Examples

{:ok, batch} =
  Kathikon.Batch.start(parent_job_id, [
    {ProcessRowWorker, %{"row" => 1}, []},
    {ProcessRowWorker, %{"row" => 2}, []}
  ], on_complete: {SummarizeWorker, %{}})

status(batch_id)

@spec status(String.t()) :: {:ok, map()} | {:error, term()}

Returns batch status by batch id.

Examples

{:ok, batch} = Kathikon.Batch.status(batch_id)
batch.status
#=> :running