Example of running an idempotent pipeline using Pydantic validation for control flow (using `OnErrorOmit`)

run.py

from pydantic import BaseModel, FilePath, NewPath, OnErrorOmit


class Step(BaseModel):
    source: FilePath
    dest: NewPath


class Pipeline(BaseModel):
    steps: list[OnErrorOmit[Step]]


def run_pipeline():
    steps = [dict(source=f"{letter}.in", dest=f"{letter}.out") for letter in "ab"]

    for step in steps:
        print(f"Prepared step: {step}")

    pipeline = Pipeline(steps=steps)

    for step in pipeline.steps:
        print(f"Running {step=}")

louis 🌟 ~/lab/pydantic/idem $ ls
run.py

louis 🌟 ~/lab/pydantic/idem $ qp -c "import run; run.run_pipeline()"
Prepared step: {'source': 'a.in', 'dest': 'a.out'}
Prepared step: {'source': 'b.in', 'dest': 'b.out'}

louis 🌟 ~/lab/pydantic/idem $ touch a.in

louis 🌟 ~/lab/pydantic/idem $ qp -c "import run; run.run_pipeline()"
Prepared step: {'source': 'a.in', 'dest': 'a.out'}
Prepared step: {'source': 'b.in', 'dest': 'b.out'}
Running step=Step(source=PosixPath('a.in'), dest=PosixPath('a.out'))

louis 🌟 ~/lab/pydantic/idem $ touch b.in

louis 🌟 ~/lab/pydantic/idem $ qp -c "import run; run.run_pipeline()"
Prepared step: {'source': 'a.in', 'dest': 'a.out'}
Prepared step: {'source': 'b.in', 'dest': 'b.out'}
Running step=Step(source=PosixPath('a.in'), dest=PosixPath('a.out'))
Running step=Step(source=PosixPath('b.in'), dest=PosixPath('b.out'))