Configurable Inventory Snapshot

Create a two-stage daily workflow that fetches order data and writes an aggregate manifest. The workflow must expose typed parameters that appear as CLI flags (--source-dir, --process-date, --stores, --min-units) generated by the dependency and honor defaults pulled from an inventory.cfg config file when provided. CLI values must take precedence over config values.

Capabilities

Config-backed defaults

• When the workflow is run with inventory.cfg provided as the config file containing:

  [InventorySnapshot]
  source_dir=/data/orders
  process_date=2024-03-02
  stores=us,eu

  running the workflow without CLI overrides writes a manifest whose fields echo source_dir=/data/orders, process_date=2024-03-02, and stores=["us","eu"]. @test

CLI overrides

• Given the above config file, invoking the workflow with --process-date 2024-03-10 --stores apac --min-units 25 writes a manifest showing process_date=2024-03-10, stores=["apac"], and min_units=25, proving CLI values win over config. @test

Typed parameters

• Supplying --process-date 2024-03-15 stores the date as ISO 2024-03-15, and invalid formats such as 15-03-2024 are rejected with a clear error. @test

Parameter propagation

• The fetch stage receives the resolved process_date, source_dir, and stores automatically from the aggregation stage (no manual argument plumbing), and the manifest echoes the values read from the upstream output. @test

Implementation

@generates

API

class LoadDailyOrders:
    """Fetches daily order data into a local target derived from the resolved parameters."""
    def output(self): ...
    def run(self): ...

class InventorySnapshot:
    """Aggregates fetched orders, filters by store list and minimum units, and writes a manifest capturing resolved parameters."""
    def requires(self): ...
    def output(self): ...
    def run(self): ...

def main(argv: list[str] | None = None) -> int:
    """Entrypoint that exposes parameterized CLI/config integration for the workflow and returns a process-style status code."""

Dependencies { .dependencies }

luigi { .dependency }

Provides workflow orchestration with parameterized tasks, CLI flags, and config file resolution.