Dependency-Driven Metric Report

Design a file-based workflow that derives summary metrics from numeric input by composing multiple dependent stages.

Capabilities

Build dependency graph for numeric pipeline

• With numbers [5, -2, 5, 10, 0], running the workflow writes a seeded input file, a cleaned numbers file containing unique non-negative integers sorted ascending ([5, 10] as JSON), a metrics file with count, sum, and average for the cleaned numbers ({"count": 2, "sum": 15, "average": 7.5}), and a final report JSON combining the cleaned numbers and metrics under "numbers" and "metrics" keys. @test

Avoid redundant work on reruns

• Running the workflow twice with the same inputs and working directory leaves the existing cleaned numbers and metrics outputs untouched (their modification times do not change on the second run). @test

Return artifact locations

• run_workflow returns absolute paths for "seed", "clean", "metrics", and "report" entries that point to the files produced during execution. @test

Implementation

@generates

API

from pathlib import Path
from typing import Iterable, Mapping

def run_workflow(numbers: Iterable[int], work_dir: Path) -> Mapping[str, Path]:
    """
    Orchestrates a multi-stage pipeline that writes:
    - a seeded input file with the provided numbers,
    - a cleaned numbers file containing only unique, non-negative integers sorted ascending,
    - a metrics file with count, sum, and average for the cleaned numbers,
    - a final report combining cleaned numbers and metrics.

    The work_dir is used for all task outputs. Returns mapping keys:
    \"seed\", \"clean\", \"metrics\", \"report\" pointing to the produced files.
    """
    ...

Dependencies { .dependencies }

luigi { .dependency }

Provides task definitions, scheduling, and dependency resolution for the workflow.