Chapter 8 · Integrations / Automation Engineer

Step pipelines with dependencies

pipeline · dependency order · step chainingAUT-40657 of 84

Real workflows aren't a flat list of independent tasks — the steps depend on each other. You can't deploy before the build passes; you can't write the report before the data loads; transform needs fetch's output. The order matters, and each step feeds the next.

A step pipeline models this. Each step has a name and a function. The engine runs them in order, passing each step's output as the input to the next — the data flows down the chain. The first step gets the initial input; the last step's output is the final result.

The key idea is chaining: a current value that each step transforms in turn.

``current = initial for step in pipeline: current = step['fn'](current) # output becomes next input``

For correctness you also want to validate dependencies: if a step declares depends_on=['load'], then load must appear *earlier* in the pipeline. A dependency that's missing entirely, or that comes *later* (a forward reference), is a broken pipeline — catch it before you run, not halfway through.

The check is a running seen-set: walk the steps in order, and a step's dependencies must all already be in seen. Add the step's own name to seen only *after* checking it — that's what catches forward references.

Syntax

# Chain outputs through the steps
current = initial_input
for step in pipeline:
    current = step['fn'](current)   # previous output → next input

# Validate dependency order
seen = set()
for step in pipeline:
    for dep in step['depends_on']:
        if dep not in seen:         # missing or forward-referenced
            return False
    seen.add(step['name'])          # add AFTER checking

Worked examples

Each step's output feeds the next

def add_step(pipeline, name, fn):
    pipeline.append({'name': name, 'fn': fn, 'depends_on': []})

def run_pipeline(pipeline, initial_input):
    current = initial_input
    steps_run = []
    for step in pipeline:
        current = step['fn'](current)
        steps_run.append(step['name'])
    return {'steps_run': steps_run, 'final_output': current}

p = []
add_step(p, 'double', lambda x: x * 2)
add_step(p, 'add_ten', lambda x: x + 10)
print(run_pipeline(p, 5))
# Output:
# {'steps_run': ['double', 'add_ten'], 'final_output': 20}

A realistic load → transform → summarise chain

def run_pipeline(steps, initial):
    current = initial
    for fn in steps:
        current = fn(current)
    return current

load      = lambda src: [10, 20, 30]
transform = lambda rows: [r + 1 for r in rows]
summarise = lambda rows: {'count': len(rows), 'total': sum(rows)}

print(run_pipeline([load, transform, summarise], 'db://prod'))
# Output:
# {'count': 3, 'total': 63}

Validating dependency order

def validate(pipeline):
    seen = set()
    for step in pipeline:
        for dep in step['depends_on']:
            if dep not in seen:
                return False
        seen.add(step['name'])
    return True

def s(name, deps):
    return {'name': name, 'depends_on': deps}

good = [s('load', []), s('transform', ['load']), s('export', ['load', 'transform'])]
bad  = [s('transform', ['load']), s('load', [])]   # forward reference
print(validate(good))   # True
print(validate(bad))    # False
# Output:
# True
# False

Common mistakes & gotchas

A step that returns None breaks the chain

If a step forgets to return (or returns None), the *next* step receives None as its input and usually crashes with a TypeError or AttributeError. Every step in a chained pipeline must return the value it wants to pass on.

Adding the name to seen before checking deps

In validate_dependencies, if you seen.add(step['name']) *before* checking that step's own depends_on, a step could 'satisfy' its own dependency and forward references slip through. Check the deps first, add the name after.

Order of registration is the order of execution

This simple engine runs steps in the order they were added — it does not topologically sort by depends_on. depends_on only *validates* the order you declared; it doesn't reorder for you. Add the steps in dependency order yourself, and use validate to confirm you got it right.

Why it matters

ETL jobs, CI/CD pipelines, deploy scripts, data workflows — they're all chains of dependent steps where the output of one is the input of the next. Modelling that explicitly, with the order validated up front, is what separates a reliable pipeline from a pile of scripts you have to remember to run in the right order by hand.