ALLM.Pipeline.Dsl (allm_pipeline v0.1.0)

Copy Markdown View Source

The construct macros behind use ALLM.Pipeline, plus the compile-time validators.

Imported into a pipeline module by ALLM.Pipeline.__using__/1; you do not use this module directly. What the constructs MEAN at runtime is documented on ALLM.Pipeline; this module is the compiler half.

Why AST, not terms

A hook cannot be stored as a value. Module.put_attribute/3 rejects anonymous functions, and &local_fun/2 written in a module body requires the function to be defined already — which a declaration block at the top of a module never satisfies. So each construct accumulates a specification containing quoted AST onto @allm_pipeline_stages, and ALLM.Pipeline.__before_compile__/1 splices that AST into the body of the generated __pipeline__(:stages), where every def/defp in the module is already defined. That is what lets a hook be written as a bare atom naming a private function (Phase 4 D6).

Hook forms

In every hook position, a bare atom is the recommended form and expands to a local capture at the declared arity; anything else — an fn, & &1.field, &Mod.fun/1 — is spliced verbatim.

Every hook's arity is __hook_arities__/0 and is not restated here: that one attribute is what the capture is built at AND what __assert_hooks_defined__!/2 checks the module defines, so a table repeating it would be a second shape of the same rule with nothing linking them.

HookArguments
metadata:(opts)
complete_metadata:(acc)
init:()
input:(ctx, subject)
over:(previous_stage_output)
body: / escape-hatch stage(ctx, subject)
skip_when:(ctx) — or the data form {:opt, key, default}
metrics …, from:(summary)
summarize(acc, ctx)
resource …, start:(ctx)
resource …, stop:(handle)
dry_run:(ctx)

subject is the previous stage's output for a stage, and the item for a fan_out. Note carry: does not capture from the subject — see ALLM.Pipeline.Dsl.Stage's carry row.

The two stage/3 forms are told apart by AST SHAPE, not by a keyword

An alias — MeetingListScraper — arrives as {:__aliases__, _, _} and means the Step form. Anything else means the escape hatch. Matching on is_atom/1 instead would treat every Step module as a hook, because an alias is an atom once expanded, and the failure would be a FunctionClauseError at the first run rather than a compile error.

Summary

Types

A validated use ALLM.Pipeline declaration. Hook values are quoted AST.

One accumulated resource declaration. Hook values are quoted AST.

One accumulated stage specification. Hook values are quoted AST.

Functions

Declare a stage that runs once per item produced by its over: hook.

Declare the one ALLM.Pipeline.Metrics row this pipeline records.

Declare a handle acquired once per run and released before the terminal write.

Declare a stage that runs once.

Declare the hook that turns the accumulator into this pipeline's summary.

Types

declaration()

@type declaration() :: %{
  name: String.t(),
  metadata: Macro.t() | nil,
  complete_metadata: Macro.t() | nil,
  init: Macro.t() | nil,
  returns: :summary | :run,
  concurrency: Macro.t(),
  borrowed_run: boolean(),
  dry_run: Macro.t() | nil,
  summary_type: atom() | nil,
  atom_hooks: [{atom(), atom(), arity()}]
}

A validated use ALLM.Pipeline declaration. Hook values are quoted AST.

resource_spec()

@type resource_spec() :: %{
  name: atom(),
  start: Macro.t(),
  stop: Macro.t(),
  atom_hooks: [{atom(), atom(), arity()}]
}

One accumulated resource declaration. Hook values are quoted AST.

stage_spec()

@type stage_spec() :: %{
  name: atom(),
  kind: :stage | :fan_out,
  step: Macro.t() | nil,
  hooks: keyword(Macro.t()),
  scalars: keyword(Macro.t()),
  atom_hooks: [{atom(), atom(), arity()}]
}

One accumulated stage specification. Hook values are quoted AST.

Functions

fan_out(name, step, opts \\ [])

(macro)

Declare a stage that runs once per item produced by its over: hook.

fan_out :detail, CommitteeDetailScraper, over: :committees_from, input: :detail_input

A fan_out always names a Step-module target. The body:-mode form was removed in Phase 4.5 — to fold an ordinary body over the items, call ALLM.Pipeline.FanOut.reduce/5 from a plain stage body instead.

Options: input: (required), over: (required), plus skip_when:, parent: and concurrency:. Not carry: — a fan-out captures nothing to propagate (removed in Phase 4.5.3; read per-item values off the [Item.t()] output instead). Not on_error: — that governs a whole-stage failure, which a fan-out does not have; both are a compile error here.

metrics(entity_type, opts)

(macro)

Declare the one ALLM.Pipeline.Metrics row this pipeline records.

metrics "meetings", from: :funnel

from: receives the value summarize produced and returns an ALLM.Pipeline.Metrics.funnel(). A pipeline that deliberately records no metrics simply omits the declaration.

resource(name, opts)

(macro)

Declare a handle acquired once per run and released before the terminal write.

resource :browser, start: :open_browser, stop: :close_browser

start: is (ctx) and returns the handle; stop: is (handle). The handle reaches every step and body as ALLM.Pipeline.Context.resource(ctx, :browser) — a struct field, never an opts key. Multiple resource declarations are acquired in declaration order and released in reverse.

Teardown ordering, its failure handling, and why a teardown failure never changes the run's status are ALLM.Pipeline.Dsl.Resource's moduledoc (Phase 4 D3).

stage(name, target, opts \\ [])

(macro)

Declare a stage that runs once.

stage :list, MeetingListScraper, input: :build_list_input
stage :committee_cache, fn _ctx, _prev -> {:ok, ensure_cache()} end

Options: input:, skip_when:, carry:, on_error:.

summarize(hook)

(macro)

Declare the hook that turns the accumulator into this pipeline's summary.

summarize :finalize          # (acc, ctx) -> term()

It runs at step 4 — before the terminal write — which is why a pipeline needing the completed %PipelineRun{} as its return value declares returns: :run instead of trying to produce it here.