Belay.Workflow (Belay v1.0.0)

Copy Markdown View Source

Jobs composed with directed acyclic dependencies.

alias Belay.Workflow

{:ok, jobs} =
  Workflow.new()
  |> Workflow.add(:fetch, MyApp.Fetch.new(%{"id" => 1}))
  |> Workflow.add(:parse, MyApp.Parse.new(%{}), deps: [:fetch])
  |> Workflow.add(:store, MyApp.Store.new(%{}), deps: [:parse])
  |> Workflow.insert(MyBelay)

Dependent jobs are inserted held and released transactionally as their upstreams succeed. When an upstream fails or is cancelled, dependents cancel by default; ignore: [:failed] / ignore: [:cancelled] (per job or workflow-wide) treats that outcome as satisfied instead.

Summary

Types

status()
t()

Functions

add(workflow, name, arg, wf_opts \\ [])

Add a named job. deps must reference previously added names.

insert(workflow, name)

Insert all workflow jobs atomically. Returns {:ok, %{name => job}}.

jobs(name, workflow_id)

All jobs in a workflow.

new(opts \\ [])

Create a workflow. Options: :workflow_id, :ignore ([:failed | :cancelled]).

status(name, workflow_id)

Status summary: %{total:, state_counts:, done?:}.

Types

status()

@type status() :: %{
  total: non_neg_integer(),
  state_counts: %{optional(String.t()) => non_neg_integer()},
  done?: boolean()
}

t()

@type t() :: %Belay.Workflow{
  entries: [map()],
  id: String.t(),
  ignore: [String.t()],
  names: MapSet.t(String.t())
}

Functions

add(workflow, name, arg, wf_opts \\ [])

@spec add(t(), String.t() | atom(), Belay.buildable(), keyword()) :: t()

Add a named job. deps must reference previously added names.

insert(workflow, name)

@spec insert(t(), Belay.instance()) :: {:ok, %{required(String.t()) => Belay.Job.t()}}

Insert all workflow jobs atomically. Returns {:ok, %{name => job}}.

jobs(name, workflow_id)

@spec jobs(Belay.instance(), String.t()) :: [Belay.Job.t()]

All jobs in a workflow.

new(opts \\ [])

@spec new(keyword()) :: t()

Create a workflow. Options: :workflow_id, :ignore ([:failed | :cancelled]).

status(name, workflow_id)

@spec status(Belay.instance(), String.t()) :: status()

Status summary: %{total:, state_counts:, done?:}.