A typed ETL pipeline

A typed ETL pipeline: extract -> transform -> load, each stage a capability with a declared contract, composed into one capability via the registry. Bad data doesn't flow downstream - it's stopped at the boundary that first notices, with every violation named.

source on github

a real captured run

PIPELINE RUN (4 raw events)

  POSTED   ev-1
  POSTED   ev-2
  REJECTED ev-3 at the 'transform' outputs boundary:
             user: is missing
             amount_cents: must be Numeric
  POSTED   ev-4

LEDGER (only facts made it this far):
  USD    1041.00
  EUR      18.50

source

# frozen_string_literal: true

# A typed ETL pipeline: extract -> transform -> load, each stage a
# capability with a declared contract, composed into one capability via
# the registry. Bad data doesn't flow downstream - it's stopped at the
# boundary that first notices, with every violation named.
#
#   bundle exec ruby examples/typed_pipeline.rb
#
# Runs offline. The interesting output is the FAILED record: watch
# where it stops and what the error says.

require class="s">"bundler/setup"
require class="s">"agentic"

registry = Agentic:class="y">:AgentCapabilityRegistry.instance

RAW_EVENTS = [
  %(id=ev-1|user=ada@example.com|amount_cents=4200|currency=USD),
  %(id=ev-2|user=grace@example.com|amount_cents=1850|currency=EUR),
  %(id=ev-3|user=|amount_cents=not-a-number|currency=USD),
  %(id=ev-4|user=joan@example.com|amount_cents=99900|currency=USD)
].freeze

def capability(name, inputs:, outputs:, &impl)
  spec = Agentic:class="y">:CapabilitySpecification.new(
    name: name, description: name, version: class="s">"1.0.0",
    inputs: inputs, outputs: outputs
  )
  Agentic.register_capability(
    spec, Agentic:class="y">:CapabilityProvider.new(capability: spec, implementation: impl)
  )
end

# Extract: raw line in, loosely-typed fields out. Extraction is
# forgiving - its job is parsing, not judgment.
capability(class="s">"extract",
  inputs: {raw: {type: class="s">"string", required: true}},
  outputs: {fields: {type: class="s">"object", required: true}}) do |input|
  fields = input[class="y">:raw].split(class="s">"|").to_h { |pair| pair.split(class="s">"=", 2) }
  {fields: fields}
end

# Transform: loose fields in, STRICT record out. This is the boundary
# where "data" becomes "facts" - the contract insists amount is a
# number and user is present.
capability(class="s">"transform",
  inputs: {fields: {type: class="s">"object", required: true}},
  outputs: {
    id: {type: class="s">"string", required: true},
    user: {type: class="s">"string", required: true},
    amount_cents: {type: class="s">"number", required: true},
    currency: {type: class="s">"string", required: true}
  }) do |input|
  fields = input[class="y">:fields]
  amount = begin
    Integer(fields[class="s">"amount_cents"])
  rescue ArgumentError, TypeError
    fields[class="s">"amount_cents"] # let the output contract catch it, by name
  end
  user = fields[class="s">"user"].to_s.empty? ? nil : fields[class="s">"user"]
  {id: fields[class="s">"id"], user: user, amount_cents: amount, currency: fields[class="s">"currency"]}.compact
end

# Load: strict record in, ledger entry out
LEDGER = Hash.new(0)
capability(class="s">"load",
  inputs: {
    id: {type: class="s">"string", required: true},
    user: {type: class="s">"string", required: true},
    amount_cents: {type: class="s">"number", required: true},
    currency: {type: class="s">"string", required: true}
  },
  outputs: {posted: {type: class="s">"string", required: true}}) do |record|
  LEDGER[record[class="y">:currency]] += record[class="y">:amount_cents]
  {posted: record[class="y">:id]}
end

# Compose the three stages into one pipeline capability
registry.compose(
  class="s">"etl_pipeline",
  class="s">"Extract, transform, and load one raw event",
  class="s">"1.0.0",
  [{name: class="s">"extract", version: class="s">"1.0.0"},
    {name: class="s">"transform", version: class="s">"1.0.0"},
    {name: class="s">"load", version: class="s">"1.0.0"}],
  lambda do |providers, inputs|
    extract, transform, load = providers
    record = transform.execute(extract.execute(raw: inputs[class="y">:raw]))
    load.execute(record)
  end
)

pipeline = registry.get_provider(class="s">"etl_pipeline")

puts class="s">"PIPELINE RUN (#{RAW_EVENTS.size} raw events)"
puts
RAW_EVENTS.each do |raw|
  posted = pipeline.execute(raw: raw)
  puts class="s">"  POSTED   #{posted[class="y">:posted]}"
rescue Agentic:class="y">:Errors:class="y">:ValidationError => e
  puts class="s">"  REJECTED #{raw[/id=([^|]+)/, 1]} at the '#{e.capability}' #{e.kind} boundary:"
  e.violations.each { |key, messages| puts class="s">"             #{key}: #{Array(messages).join(", class="s">")}" }
end

puts
puts class="s">"LEDGER (only facts made it this far):"
LEDGER.each { |currency, cents| puts format(class="s">"  %s %10.2f", currency, cents / 100.0) }