Inputs

Feed values into a graph with InputParam (a value given at execute()) or a source (a file or folder whose content is part of the cache key).

• InputParam[T]("name"): required; pass name=... to execute(). Add value_type=T so the CLI parses --param as T.
• InputParam[T].with_default("name", value): optional.
• FileSource(path): the file's text. Its content is hashed, so editing the file recomputes.
• LinesSource(path), JSONLSource(path): the file as lines, or as parsed JSON lines.
• DirSource(path, pattern): a folder; adding, removing or editing a matching file recomputes.
• A FileSource, LinesSource or JSONLSource on a directory raises IsADirectoryError. The message names the source and the path. Use DirSource for a folder.
• A file that does not exist: the run fails with FileNotFoundError when the source is read.
• DirSource uses pattern as a glob. The default "*.jsonl" is not recursive: pass "**/*.jsonl" for nested folders.

"""Inputs: InputParam values and FileSource content are part of the cache key."""

import tempfile
from dataclasses import dataclass
from pathlib import Path

from pydantic import BaseModel

from adag import FileSource, InputParam, NullObserver, Pipeline, Step


class Count(BaseModel):
    words: int


@dataclass(frozen=True)
class CountWords(Step[Count]):
    def run(self, text: str, minimum: int) -> Count:
        return Count(words=sum(1 for word in text.split() if len(word) >= minimum))


with tempfile.TemporaryDirectory() as tmp:
    data = Path(tmp) / "notes.txt"
    data.write_text("a short note about caching")

    minimum = InputParam[int].with_default("minimum", 1)
    graph = CountWords()(text=FileSource(str(data)), minimum=minimum)
    pipeline = Pipeline(graph, cache_dir=Path(tmp) / "cache")

    print(pipeline.execute(observer=NullObserver()))
    print(pipeline.execute(observer=NullObserver(), minimum=5))
    data.write_text("the file changed, so the key changed too")
    print(pipeline.execute(observer=NullObserver()))

words=5
words=3
words=8