API Reference

This page provides a detailed API reference for NovaPipe modules, classes, and functions.
Powered by mkdocstrings, each section below will automatically list available attributes and docstrings.

---

Core Modules

novapipe.tasks

Discover and register task functions for pipelines.

aggregate_results(params)

Stub aggregate_results: just return whatever was passed in.

Source code in novapipe/tasks.py

@task
def aggregate_results(params: Dict[str, Any]) -> Any:
    """
    Stub aggregate_results: just return whatever was passed in.
    """
    # If you passed a dict of results, return that; otherwise echo params
    return params

async_wait_and_print(params) async

An example async task that waits N seconds, then prints a message. Pipeline usage:

tasks

• name: delayed task: async_wait_and_print params: message: "This ran after waiting" seconds: 2

Source code in novapipe/tasks.py

@task
async def async_wait_and_print(params: Dict) -> None:
    """
    An example async task that waits N seconds, then prints a message.
    Pipeline usage:

    tasks:
      - name: delayed
        task: async_wait_and_print
        params:
          message: "This ran after waiting"
          seconds: 2
    """
    seconds = params.get("seconds", 1)
    message = params.get("message", "")
    # Simple asyncio sleep
    import asyncio

    await asyncio.sleep(seconds)
    print(message)

call_api(params)

Stub call_api: pretend to fetch from url by returning a marker string.

Source code in novapipe/tasks.py

@task
def call_api(params: Dict[str, Any]) -> Any:
    """
    Stub call_api: pretend to fetch from `url` by returning a marker string.
    """
    url = params.get("url", "")
    return f"fetched:{url}"

count_file_lines(params)

Count the number of lines in the file at path. Return the integer count.

Source code in novapipe/tasks.py

@task
def count_file_lines(params: Dict) -> int:
    """
    Count the number of lines in the file at `path`. Return the integer count.
    """
    path = params.get("path")
    if not path or not os.path.isfile(path):
        raise RuntimeError(f"File not found: {path!r}")
    with open(path) as f:
        return sum(1 for _ in f)

create_temp_dir(params)

Create a unique temporary directory under base. Return its path.

Source code in novapipe/tasks.py

@task
def create_temp_dir(params: Dict) -> str:
    """
    Create a unique temporary directory under `base`. Return its path.
    """
    base = params.get("base", None)
    if base and not os.path.isdir(base):
        raise RuntimeError(f"Base directory {base!r} does not exist.")
    tmpdir = tempfile.mkdtemp(prefix="novapipe_", dir=base)
    return tmpdir

echo(params)

Print and return params['message'].

Source code in novapipe/tasks.py

@task
def echo(params: Dict) -> Any:
    """
    Print and return params['message'].
    """
    msg = params.get("message", "")
    print(msg)
    return msg

extract_data(params)

Stub extract_data: returns the source param so we can see it ran.

Source code in novapipe/tasks.py

@task
def extract_data(params: Dict[str, Any]) -> Any:
    """
    Stub extract_data: returns the `source` param so we can see it ran.
    """
    return params.get("source")

load_data(params)

Stub load_data: prints and returns a load message.

Source code in novapipe/tasks.py

@task
def load_data(params: Dict[str, Any]) -> str:
    """
    Stub load_data: prints and returns a load message.
    """
    msg = f"loaded: {params}"
    print(msg)
    return msg

load_plugins()

Discover and load external plugins via entry point group 'novapipe.plugins'. Plugins should define entry_points in their own pyproject.

Source code in novapipe/tasks.py

def load_plugins() -> None:
    """
    Discover and load external plugins via entry point group 'novapipe.plugins'.
    Plugins should define entry_points in their own pyproject.
    """

    # Gather all (dist_name, dist_version, entry_point) tuples
    ep_map: Dict[str, list[tuple[str, str, EntryPoint]]] = {}
    for dist in distributions():
        dist_name = dist.metadata.get("Name", dist.name)
        dist_version = dist.version
        for ep in dist.entry_points:
            if ep.group == "novapipe.plugins":
                ep_map.setdefault(ep.name, []).append((dist_name, dist_version, ep))

    # Resolve and register
    errors: list[str] = []
    for task_name, candidates in ep_map.items():
        if len(candidates) == 1:
            dist_name, dist_version, ep = candidates[0]
        else:
            # multiple candidates -> need pin
            # find matches among candidates
            matched = [
                (d, v, e) for (d, v, e) in candidates
                if _plugin_pins.get(d) == v
            ]
            if len(matched) == 1:
                dist_name, dist_version, ep = matched[0]
            else:
                # build a helpful error message
                opts = ", ".join(f"{d}=={v}" for d, v, _ in candidates)
                errors.append(
                    f"Task '{task_name}' is provided by multiple plugins: {opts}. "
                    f"Pin one with --plugin-version DIST==VERSION."
                )
                continue

        # load and register
        func = ep.load()
        if task_name in task_registry:
            # overriding a built-in or earlier plugin; warn or allow if pinned
            logging.getLogger("novapipe").warning(
                f"Task '{task_name}' from plugin {dist_name}=={dist_version} "
                f"is overriding existing registration"
            )
        task_registry[task_name] = func

    if errors:
        raise RuntimeError("Plugin load errors:\n  " + "\n  ".join(errors))

maybe_fail(params)

A demo task that randomly raises to simulate flakiness.

Source code in novapipe/tasks.py

@task
def maybe_fail(params: Dict) -> None:
    """
    A demo task that randomly raises to simulate flakiness.
    """
    attempt_id = params.get("attempt_id", None)
    if random.random() < 0.5:  # 50% chance to fail
        raise RuntimeError(f"Simulated failure for attempt_id={attempt_id}")
    print(f"✅ maybe_fail succeeded (attempt_id={attempt_id})")

print_message(params)

A simple built-in task that prints the "message" field from params. Example usage in pipeline.yaml:

tasks

• name: say_hello task: print_message params: message: "Hello, NovaPipe!"

Source code in novapipe/tasks.py

@task
def print_message(params: Dict) -> None:
    """
    A simple built-in task that prints the "message" field from params.
    Example usage in pipeline.yaml:

    tasks:
      - name: say_hello
        task: print_message
        params:
          message: "Hello, NovaPipe!"
    """
    msg = params.get("message", "")
    print(msg)

return_value(params)

Return whatever is in params['value'].

Source code in novapipe/tasks.py

@task
def return_value(params: Dict) -> Any:
    """
    Return whatever is in params['value'].
    """
    return params.get("value")

set_plugin_pins(pins)

pins: mapping of distribution name -> version to pin for that plugin. Example: {'novapipe-foo': '0.2.1'}

Source code in novapipe/tasks.py

def set_plugin_pins(pins: Dict[str, str]) -> None:
    """
    pins: mapping of distribution name -> version to pin for that plugin.
    Example: {'novapipe-foo': '0.2.1'}
    """
    global _plugin_pins
    _plugin_pins = pins

task(func)

Decorator to register a function (sync or async) as a NovaPipe task. Usage: @task() def my_task(param1, param2): ...

Source code in novapipe/tasks.py

def task(func: Callable[[dict], None]) -> Callable[[dict], None]:
    """
    Decorator to register a function (sync or async) as a NovaPipe task.
    Usage:
    @task()
    def my_task(param1, param2):
        ...
    """
    task_registry[func.__name__] = func
    return func

transform_data(params)

Stub transform_data: tags the data as transformed.

Source code in novapipe/tasks.py

@task
def transform_data(params: Dict[str, Any]) -> Any:
    """
    Stub transform_data: tags the data as transformed.
    """
    src = params.get("source") or params.get("extracted")
    return f"transformed({src})"

upload_file_s3(params)

Uploads a local file at path to S3 bucket/key, returns the S3 URI.

Source code in novapipe/tasks.py

@task
def upload_file_s3(params: Dict[str, str]) -> str:
    """
    Uploads a local file at `path` to S3 bucket/key, returns the S3 URI.
    """
    s3 = boto3.client("s3")
    bucket = params["bucket"]
    key = params["key"]
    path = params["path"]
    s3.upload_file(path, bucket, key)
    return f"s3://{bucket}/{key}"

wrap_text(params)

Return 'WRAPPED: ' where input = params['input'].

Source code in novapipe/tasks.py

@task
def wrap_text(params: Dict) -> str:
    """
    Return 'WRAPPED: <input!r>' where input = params['input'].
    """
    inp = params.get("input", "")
    return f"WRAPPED: {inp!r}"

write_text_file(params)

Create a text file at path with content content. Return the file path.

Source code in novapipe/tasks.py

@task
def write_text_file(params: Dict) -> str:
    """
    Create a text file at `path` with content `content`. Return the file path.
    """
    path = params.get("path")
    content = params.get("content", "")
    if not path:
        raise RuntimeError("Missing 'path' in params for write_text_file.")
    # Ensure directory exists
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w") as f:
        f.write(content)
    return path

---

novapipe.cli

The command-line interface definitions, including commands, options, and subcommands.

cli(verbose)

NovaPipe — lightweight, plugin-driven ETL orchestration.

Source code in novapipe/cli.py

@click.group()
@click.version_option("0.1.0", package_name="novapipe", prog_name="novapipe")
@click.option(
    "--verbose", "-v", is_flag=True, default=False, help="Enable DEBUG logging for NovaPipe."
)
def cli(verbose) -> None:
    """
     NovaPipe — lightweight, plugin-driven ETL orchestration.
     """
    level = "DEBUG" if verbose else "INFO"
    configure_logging(level=level)

dag(pipeline_file, export_dot)

Show task-dependency graph for a pipeline. By default, prints an ASCII view. Use --dot to emit Graphviz DOT.

Source code in novapipe/cli.py

@cli.command("dag")
@click.argument("pipeline_file", type=click.Path(exists=True))
@click.option(
    "--dot",
    "export_dot",
    is_flag=True,
    default=False,
    help="Output Graphviz DOT instead of ASCII.",
)
def dag(pipeline_file, export_dot):
    """
    Show task-dependency graph for a pipeline.
    By default, prints an ASCII view. Use --dot to emit Graphviz DOT.
    """
    import yaml

    with open(pipeline_file) as f:
        data = yaml.safe_load(f)

    # Initialize runner (validates & builds graph)
    runner = PipelineRunner(data)

    if export_dot:
        dot_text = runner.to_dot()
        click.echo(dot_text)
    else:
        click.echo("🚀 NovaPipe DAG:")
        runner.print_dag()

describe(task_name)

Show the full signature and docstring of a given task.

Source code in novapipe/cli.py

@cli.command("describe")
@click.argument("task_name")
def describe(task_name: str):
    """
    Show the full signature and docstring of a given task.
    """
    load_plugins()
    func = task_registry.get(task_name)
    if not func:
        click.echo(f"❌ Task {task_name!r} not found.", err=True)
        raise SystemExit(1)

    # ---- Gather plugin metadata ----
    plugin_info = {}
    for dist in distributions():
        for ep in dist.entry_points:
            if ep.group == "novapipe.plugins":
                module_name = ep.value
                # Load the plugin module and scan for tasks
                try:
                    mod = __import__(module_name, fromlist=["*"])
                except ImportError:
                    continue
                for fname, fobj in vars(mod).items():
                    if callable(fobj) and getattr(fobj, "__wrapped__", None):
                        # decorated with @task
                        plugin_info[fname] = {
                            "distribution": dist.metadata.get("Name", dist.name),
                            "version": dist.version,
                        }

    # Signature
    sig = _inspect.signature(func)
    click.echo(f"{task_name}{sig}\n")

    # Docstring (or placeholder)
    doc = _inspect.getdoc(func) or "(no documentation provided)"

    # Render as Markdown if Rich is available, else plain text
    try:
        from rich.console import Console
        from rich.markdown import Markdown

        console = Console()
        console.print(Markdown(doc))
    except ImportError:
        click.echo(doc)

    # If this is a plugin task, show where it came from
    info = plugin_info.get(task_name)
    if info:
        click.echo()
        click.echo("Plugin Metadata:")
        click.echo(f" • Distribution: {info['distribution']} (v{info['version']})")
        click.echo(f" • Module:       {func.__module__}")

init(name)

Create a started pipeline YAML file.

Source code in novapipe/cli.py

@cli.command()
@click.argument("name", required=False, default="pipeline.yaml")
def init(name) -> None:
    """
    Create a started pipeline YAML file.
    """
    template = """\
# NovaPipe pipeline template

tasks:
    - name: print_message
      params:
        message: "Hello, NovaPipe!
    """
    if os.path.exists(name):
        click.echo(f"!  {name} already exists. Aborting.")
        raise SystemExit(1)

    with open(name, 'w') as f:
        f.write(template)
    click.echo(f"Initialized pipeline template at {name}")

inspect()

List all registered tasks.

Source code in novapipe/cli.py

@cli.command()
def inspect() -> None:
    """List all registered tasks."""
    load_plugins()
    click.echo("Registered tasks:")
    for name, func in sorted(task_registry.items()):
        sig = _inspect.signature(func)
        click.echo(f"- {name}{sig}")

playground(vars, template)

Interactive Jinja2 playground

If TEMPLATE is given, renders it once and exits. Otherwise enters a REPL: type templates, ENTER to render, or 'exit' or quit.

Source code in novapipe/cli.py

@cli.command("playground")
@click.option(
    "--var", "-D",
    "vars",
    metavar="KEY=VAL",
    multiple=True,
    help="Seed the playground context with KEY=VAL (can repeat).",
)
@click.argument("template", required=False)
def playground(vars, template):
    """
    Interactive Jinja2 playground

    If TEMPLATE is given, renders it once and exits.
    Otherwise enters a REPL: type templates, ENTER to render, or 'exit' or quit.
    """
    # Build the Jinja2 env like NovaPipe uses
    env = jinja2.Environment(
        undefined=jinja2.StrictUndefined,
        autoescape=False,
    )
    env.globals.update({'int': int, 'float': float, 'str': str, 'bool': bool, 'len': len})

    # Seed context
    context = {}
    for var in vars:
        if "=" not in var:
            click.echo(f"❌ Invalid var: {var!r}. Use KEY=VAL.", err=True)
            return
        k, v = var.split("=", 1)
        # Try to parse JSON for numbers/bools/lists, else keep string
        try:
            context[k] = json.loads(v)
        except Exception:
            context[k] = v

    click.echo(f"🔧 Playground context: {context!r}")

    def render_and_print(tmpl_str: str):
        try:
            tmpl = env.from_string(tmpl_str)
            out = tmpl.render(**context)
            click.echo(out)
        except Exception as e:
            click.echo(f"⚠️  Error: {e}", err=True)

    if template:
        # One-off render
        render_and_print(template)
        return

    # REPL mode
    click.echo("📝 Enter templates; type 'exit' or Ctrl-D to quit.")
    while True:
        try:
            line = input(">>> ")
        except (EOFError, KeyboardInterrupt):
            click.echo("\n👋 Exiting playground.")
            break
        if not line or line.strip().lower() in ("exit", "quit"):
            click.echo("👋 Goodbye.")
            break
        render_and_print(line)

plugin()

Plugin management commands.

Source code in novapipe/cli.py

@cli.group()
def plugin():
    """
    Plugin management commands.
    """
    pass

plugin_ci_template(output_path)

Scaffold a GitHub Actions workflow to build, test, and publish this plugin.

Source code in novapipe/cli.py

@plugin.command("ci-template")
@click.argument(
    "output_path",
    type=click.Path(dir_okay=False, writable=True),
    default=".github/workflows/publish-plugin.yml",
)
def plugin_ci_template(output_path):
    """
    Scaffold a GitHub Actions workflow to build, test, and publish this plugin.
    """
    template = """\
name: Publish NovaPipe Plugin

on:
  push:
    tags:
      - 'v*'  # Trigger on any tag like v1.2.3

jobs:
  build-test-publish:
    runs-on: ubuntu-latest

    steps:
      - name: Checkout code
        uses: actions/checkout@v3

      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.9'  # or matrix 3.8–3.12

      - name: Install Poetry
        run: |
          pip install poetry

      - name: Configure PyPI token
        run: |
          poetry config pypi-token.pypi ${{{{ secrets.PYPI_API_TOKEN }}}}

      - name: Install dependencies
        run: |
          poetry install --no-dev

      - name: Run tests
        run: |
          poetry run pytest --cov=.

      - name: Build distribution
        run: |
          poetry build

      - name: Publish to PyPI
        run: |
          poetry publish --no-interaction --username __token__ --password ${{{{ secrets.PYPI_API_TOKEN }}}}

      - name: Create GitHub Release
        uses: ncipollo/release-action@v1
        with:
          tag: ${{{{ github.ref_name }}}}
"""
    # Ensure parent directories exist
    out_path = Path(output_path)
    out_path.parent.mkdir(parents=True, exist_ok=True)

    with open(out_path, "w") as f:
        f.write(template)
    click.echo(f"✅ CI workflow scaffolded to {output_path}")
    click.echo("👉 Next steps:")
    click.echo("   • Add PYPI_API_TOKEN as a secret in your repo settings")
    click.echo("   • Commit and push this file to `.github/workflows/`")
    click.echo("   • Tag a new release (`git tag v0.1.0 && git push --tags`) to trigger it")

plugin_list(dists, tasks_filter, sources)

List all installed NovaPipe plugins (distribution, version, module, tasks).

Source code in novapipe/cli.py

@plugin.command("list")
@click.option(
    "--dist",
    "dists",
    metavar="DIST",
    multiple=True,
    help="Only show plugins from this distribution (can repeat).",
)
@click.option(
    "--task",
    "tasks_filter",
    metavar="TASK_SUBSTR",
    multiple=True,
    help="Only show entries whose task name contains this substring (can repeat).",
)
@click.option(
    "--source",
    "sources",
    metavar="MODULE_SUBSTR",
    multiple=True,
    help="Only show entries whose module path contains this substring (can repeat).",
)
def plugin_list(dists, tasks_filter, sources):
    """
    List all installed NovaPipe plugins (distribution, version, module, tasks).
    """
    load_plugins()
    # Gather plugins by distribution
    plugins = {}
    for dist in distributions():
        name = dist.metadata.get("Name", dist.name)
        version = dist.version
        for ep in dist.entry_points:
            if ep.group == "novapipe.plugins":
                # ep.name is the entry-point alias (i.e. the task name)
                task_name = ep.name
                module = ep.value

                # Apply filters immediately
                if dists and name not in dists:
                    continue
                if tasks_filter and not any(f in task_name for f in tasks_filter):
                    continue
                if sources and not any(s in module for s in sources):
                    continue

                key = f"{name} (v{version})"
                plugins.setdefault(key, []).append((task_name, module))

    if not plugins:
        click.echo("No matching NovaPipe plugins installed.")
        return

    click.echo("Installed NovaPipe plugins:")
    for dist_key, tasks in sorted(plugins.items()):
        click.echo(f"\n• {dist_key}")
        for task_name, module in sorted(tasks):
            click.echo(f"    – {task_name}  (module: {module})")

plugin_scaffold(plugin_name)

Scaffold a new NovaPipe plugin project.

Source code in novapipe/cli.py

@plugin.command("scaffold")
@click.argument("plugin_name")
def plugin_scaffold(plugin_name: str) -> None:
    """
    Scaffold a new NovaPipe plugin project.
    """
    # e.g. plugin_name = "novapipe foo"
    project_dir = os.path.abspath(plugin_name)
    if os.path.exists(project_dir):
        click.echo(f"! Directory {project_dir} already exists. Aborting.")
        raise SystemExit(1)

    # create structure
    os.makedirs(os.path.join(project_dir, plugin_name))
    with open(os.path.join(project_dir, "pyproject.toml"), "w") as f:
        f.write(f"""\
[tool.poetry]
name = "{plugin_name}"
version = "0.1.0"
description = "A NovaPipe plugin providing additional tasks."
authors = ["<you@example.com>"]

[tool.poetry.dependencies]
python = "^3.8"
novapipe = "^0.1.0"

[tool.poetry.plugins."novapipe.plugins"]
{plugin_name} = "{plugin_name}.tasks"
""")

    with open(os.path.join(project_dir, plugin_name, "__init__.py"), "w") as f:
        f.write("# Plugin package for NovaPipe\n")

    with open(os.path.join(project_dir, plugin_name, "tasks.py"), "w") as f:
        f.write(f"""\
from novapipe.tasks import task

@task
def hello_{plugin_name}(params: dict):
    \"""
    Sample task for the {plugin_name} plugin.
    \"""
    print("Hello from {plugin_name}!", params)
""")

    with open(os.path.join(project_dir, "README.md"), "w") as f:
        f.write(f"# {plugin_name}\n\nA NovaPipe plugin scaffold by `novapipe plugin scaffold {plugin_name}`.\n")

    click.echo(f"✅ Scaffolded new plugin at {project_dir}")
    click.echo("👉 Next steps:")
    click.echo(f"   • cd {plugin_name}")
    click.echo("   • poetry install")
    click.echo("   • Implement your tasks in tasks.py")
    click.echo("   • git init && git add . && git commit -m 'Initial plugin scaffold'")

repo()

Manage CI/CD workflows for the NovaPipe repository itself.

Source code in novapipe/cli.py

@cli.group("repo")
def repo():
    """
    Manage CI/CD workflows for the NovaPipe repository itself.
    """
    pass

repo_ci_template(output_path)

Scaffold a GitHub Actions workflow for canary (TestPyPI) and stable (PyPI) releases.

Source code in novapipe/cli.py

@repo.command("ci-template")
@click.argument(
    "output_path",
    type=click.Path(dir_okay=False, writable=True),
    default=".github/workflows/release.yml",
)
def repo_ci_template(output_path):
    """
    Scaffold a GitHub Actions workflow for canary (TestPyPI) and stable (PyPI) releases.
    """
    template = """\
name: Release NovaPipe

on:
  push:
    branches:
      - main
  push:
    tags:
      - 'v*'

jobs:
  canary:
    if: github.event_name == 'push' && github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    steps:
      - name: Checkout code
        uses: actions/checkout@v3

      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.9'

      - name: Install Poetry
        run: pip install poetry

      - name: Configure TestPyPI token
        run: poetry config repositories.testpypi https://test.pypi.org/legacy/ && \\
             poetry config pypi-token.testpypi ${{{{ secrets.TESTPYPI_API_TOKEN }}}}

      - name: Install dependencies
        run: poetry install --no-dev

      - name: Bump to canary version
        # e.g. from 1.2.3 to 1.2.4-dev$(date +%Y%m%d%H%M)
        run: |
          base=$(poetry version -s)
          stamp=$(date +%Y%m%d%H%M)
          poetry version "${{ base }}-dev${{ stamp }}"

      - name: Build distribution
        run: poetry build

      - name: Publish to TestPyPI
        run: |
          poetry publish --repository testpypi --username __token__ \\
            --password ${{{{ secrets.TESTPYPI_API_TOKEN }}}}

  stable:
    if: startsWith(github.ref, 'refs/tags/v')
    runs-on: ubuntu-latest
    needs: canary
    steps:
      - name: Checkout code
        uses: actions/checkout@v3

      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.9'

      - name: Install Poetry
        run: pip install poetry

      - name: Configure PyPI token
        run: poetry config pypi-token.pypi ${{{{ secrets.PYPI_API_TOKEN }}}}

      - name: Install dependencies
        run: poetry install --no-dev

      - name: Build distribution
        run: poetry build

      - name: Publish to PyPI
        run: |
          poetry publish --no-interaction --username __token__ \\
            --password ${{{{ secrets.PYPI_API_TOKEN }}}}
"""
    out = Path(output_path)
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text(template)
    click.echo(f"✅ Repository CI workflow scaffolded to {output_path}")
    click.echo("👉 Next steps:")
    click.echo("   • Add secrets TESTPYPI_API_TOKEN & PYPI_API_TOKEN in Settings → Secrets")
    click.echo("   • Commit and push this file to .github/workflows/")
    click.echo("   • Pushing to main creates a TestPyPI canary; tagging vX.Y.Z on GitHub publishes to PyPI")

report(summary_json)

Read a summary JSON (as written by --summary-json) and print a table: name status attempts duration(s) error

Source code in novapipe/cli.py

@cli.command("report")
@click.argument(
    "summary_json",
    type=click.Path(exists=True, dir_okay=False),
)
def report(summary_json):
    """
    Read a summary JSON (as written by --summary-json) and print a table:
        name    status  attempts    duration(s)     error
    """
    # Load the JSON
    with open(summary_json) as f:
        data = json.load(f)

    tasks: List[Dict[str, Any]] = data.get("tasks", [])
    if not tasks:
        click.echo("No tasks found in summary.", err=True)
        return

    # Prepare rows
    headers = ["Name", "Status", "Attempts", "Duration(s)", "Error"]
    rows = []
    for t in tasks:
        rows.append([
            t.get("name", ""),
            t.get("status", ""),
            str(t.get("attempts", "")),
            f"{t.get('duration_secs', 0):.3f}",
            t.get("error") or "",
        ])

    # Try using tabulate if available
    try:
        from tabulate import tabulate
        table = tabulate(rows, headers=headers, tablefmt="github")
        click.echo(table)
    except ImportError:
        # Fallback to simple padding
        # compute max widths
        widths = [len(h) for h in headers]
        for row in rows:
            for i, cell in enumerate(row):
                widths[i] = max(widths[i], len(cell))

        # header line
        hdr = "  ".join(h.ljust(widths[i]) for i, h in enumerate(headers))
        click.echo(hdr)
        click.echo("-" * len(hdr))
        # rows
        for row in rows:
            line = "  ".join(row[i].ljust(widths[i]) for i in range(len(headers)))
            click.echo(line)

run(pipeline_file, vars, summary_path, metrics_port, metrics_path, plugin_versions, ignore_failures)

Run a pipeline YAML file.

Source code in novapipe/cli.py

@cli.command()
@click.argument("pipeline_file", type=click.Path(exists=True))
@click.option(
    "--var", "-D",
    "vars",
    metavar="KEY=VAL",
    multiple=True,
    help="Set a pipeline variable (can be used multiple times).",
)
@click.option(
    "--summary-json",
    "summary_path",
    type=click.Path(dir_okay=False, writable=True),
    default=None,
    help="Path to write task summary JSON after run"
)
@click.option(
    "--metrics-port",
    type=int,
    default=None,
    help="If set, start a Prometheus metrics server on this port.",
)
@click.option(
    "--metrics-path", "-p",
    type=str,
    default="/metrics",
    show_default=True,
    help="HTTP path under which to expose metrics.",
)
@click.option(
    "--ignore-failures",
    is_flag=True,
    default=False,
    help="Globally treat all tasks as ignore_failure=True (unless overridden per-task).",
)
@click.option(
    "--plugin-version",
    "plugin_versions",
    metavar="DIST==VERSION",
    multiple=True,
    help="Pin a plugin distribution to a version, e.g. novapipe-foo==0.2.1"
)
def run(pipeline_file: str, vars: list, summary_path: str, metrics_port: int, metrics_path: str,
        plugin_versions: Any, ignore_failures: bool) -> None:
    """Run a pipeline YAML file."""
    with open(pipeline_file) as f:
        data = yaml.safe_load(f)

    # Parse and set plugin-version pins before loading
    pins: Dict[str, str] = {}
    for pv in plugin_versions:
        if "==" not in pv:
            click.echo(f"❌ Invalid --plugin-version format: {pv!r}. Expected DIST==VERSION", err=True)
            raise SystemExit(1)
        dist, ver = pv.split("==", 1)
        pins[dist] = ver
    set_plugin_pins(pins)
    load_plugins()

    # Derive a pipeline name from the file, e.g. 'pipeline.yaml' -> 'pipeline'
    pipeline_name = os.path.splitext(os.path.basename(pipeline_file))[0]
    runner = PipelineRunner(data, pipeline_name=pipeline_name)

    # If global ignore-failures is set, override each task_model.ignore_failure
    if ignore_failures:
        for tm in runner.tasks_by_name.values():
            # If user explicitly set ignore_failure in YAML to False, respect that
            if not hasattr(tm, "ignore_failure") or tm.ignore_failure is False:
                tm.ignore_failure = False

    # Parse CLI vars and seed runner.context
    for var_pair in vars:
        if "=" not in var_pair:
            click.echo(f"❌ Invalid --var format: {var_pair!r}. Use KEY=VAL.", err=True)
            raise SystemExit(1)
        key, val = var_pair.split("=", 1)
        runner.context[key] = val

    start = time.time()
    try:
        # Start metrics server if requested, on the configured path
        if metrics_port:
            class _MetricsHandler(BaseHTTPRequestHandler):
                def do_GET(self):
                    if self.path == metrics_path:
                        self.send_response(200)
                        self.send_header("Content-Type", CONTENT_TYPE_LATEST)
                        self.end_headers()
                        self.wfile.write(generate_latest())
                    else:
                        self.send_response(404)

            httpd = HTTPServer(("0.0.0.0", metrics_port), _MetricsHandler)
            thread = threading.Thread(target=httpd.serve_forever, daemon=True)
            thread.start()
            click.echo(f"📊 Metrics available at http://localhost:{metrics_port}{metrics_path}")

        # Measure overall pipeline duration & status
        start = time.time()
        summary = runner.run()  # now returns PipelineRunSummary, with seeded context used in templates
        dur = time.time() - start

        # Record pipeline-level metrics
        PIPELINE_STATUS.labels(pipeline=pipeline_name, status="success").inc()
        PIPELINE_DURATION.labels(pipeline=pipeline_name).observe(dur)

        click.echo("✅ Pipeline completed (check logs for details).")

        # If we're serving metrics, keep the process alive until Ctrl+C
        if metrics_port:
            click.echo("🔒 Metrics server still running. Press Ctrl+C to stop.")
            try:
                while True:
                    time.sleep(1)
            except KeyboardInterrupt:
                click.echo("\n👋 Shutting down metrics server and exiting.")

        if summary_path:
            # Write JSON summary to disk
            import json

            out = {"tasks": summary.to_list()}
            with open(summary_path, "w") as jf:
                json.dump(out, jf, indent=2)
            click.echo(f"📝 Summary written to {summary_path}")
    except Exception as e:
        # Record pipeline failure metric
        dur = time.time() - start
        PIPELINE_STATUS.labels(pipeline=pipeline_name, status="failed").inc()
        PIPELINE_DURATION.labels(pipeline=pipeline_name).observe(dur)

        click.echo(f"❌ Pipeline failed: {e}", err=True)
        raise SystemExit(1)

tutorial(task_name)

Emit a "Try-me" pipeline YAML snippet for TASK_NAME.

Source code in novapipe/cli.py

@cli.command("tutorial")
@click.argument("task_name")
def tutorial(task_name):
    """
    Emit a "Try-me" pipeline YAML snippet for TASK_NAME.
    """
    load_plugins()
    func = task_registry.get(task_name)
    if not func:
        click.echo(f"❌ Task {task_name!r} not found.", err=True)
        raise SystemExit(1)

    doc = _inspect.getdoc(func) or ""
    # 1) Try to extract an Example: block
    # We look for lines starting with "Example:" and take the indented block after it.
    example_match = re.search(
        r"Example:\s*\n((?:[ \t]+.*\n?)+)", doc, flags=re.IGNORECASE
    )
    if example_match:
        snippet = textwrap.dedent(example_match.group(1))
        click.echo("Here's a usage example from the docstring:\n")
        click.echo(snippet.rstrip())
        return

    # 2) Fallback: minimal skeleton
    click.echo(f"# Demo pipeline for `{task_name}`\n")
    click.echo("tasks:")
    click.echo(f"  - name: example_{task_name}")
    click.echo(f"    task: {task_name}")
    click.echo("    params:")
    click.echo("      # TODO: fill in parameters for this task")
    click.echo("\n# Run with:")
    click.echo(f"#   novapipe run pipeline.yaml")

---

novapipe.models

Pydantic models defining the pipeline schema: TaskModel and Pipeline.

---

novapipe.runner

The execution engine for pipelines: PipelineRunner, RateLimiter, and helper functions.

PipelineRunSummary

Collects a dict of TaskSummary, keyed by task name.

Source code in novapipe/runner.py

class PipelineRunSummary:
    """
    Collects a dict of TaskSummary, keyed by task name.
    """
    def __init__(self):
        self.tasks: Dict[str, TaskMetrics] = {}

    def record_start(self, name: str):
        ts = TaskMetrics(name)
        self.tasks[name] = ts
        ts.start_time = time.time()  # for intermediate tracking

    def record_success(self, name: str, attempts: int):
        ts = self.tasks[name]
        ts.attempts = attempts
        ts.status = "success"
        ts.duration_secs = time.time() - ts.start_time
        ts.error = None

    def record_failed_ignored(self, name: str, attempts: int, error: Exception):
        ts = self.tasks[name]
        ts.attempts = attempts
        ts.status = "failed_ignored"
        ts.duration_secs = time.time() - ts.start_time
        ts.error = repr(error)

    def record_failed_abort(self, name: str, attempts: int, error: Exception):
        ts = self.tasks[name]
        ts.attempts = attempts
        ts.status = "failed_abort"
        ts.duration_secs = time.time() - ts.start_time
        ts.error = repr(error)

    def record_skipped(self, name: str):
        ts = TaskMetrics(name)
        ts.attempts = 0
        ts.status = "skipped"
        ts.duration_secs = 0.0
        ts.error = None
        ts.start_time = time.time()
        self.tasks[name] = ts

    def to_list(self) -> List[Dict[str, Any]]:
        return [ts.to_dict() for ts in self.tasks.values()]

PipelineRunner

Executes a validated Pipeline of Steps, supporting async tasks.

Source code in novapipe/runner.py

class PipelineRunner:
    """
    Executes a validated Pipeline of Steps, supporting async tasks.
    """
    def __init__(self, raw_data: dict, pipeline_name: str) -> None:
        # 1. Parse & validate YAML into Pydantic models
        self.pipeline = Pipeline.model_validate(raw_data)
        # 2. Build in-memory DAG structures
        self._build_graph()

        # Build rate limiters per key
        self._rate_limiters: Dict[str, RateLimiter] = {}
        for t in self.tasks_by_name.values():
            if t.rate_limit is not None:
                key = t.rate_limit_key or t.name
                if key in self._rate_limiters:
                    # pick the lowest rate if multiple tasks share the same key
                    existing = self._rate_limiters[key].rate
                    self._rate_limiters[key].rate = min(existing, t.rate_limit)
                else:
                    self._rate_limiters[key] = RateLimiter(rate=t.rate_limit, per=1.0)

        self.pipeline_name = pipeline_name

        # Prepare summary
        self._summary = PipelineRunSummary()

        # Shared context: task_name → return_value
        self.context: Dict[str, Any] = {}

        # Jinja2 environment for templating
        self._jinja_env = jinja2.Environment(
            undefined=jinja2.StrictUndefined,
            autoescape=False
        )
        # make python build-ins available in templates
        self._jinja_env.globals.update({
            'int': int,
            'float': float,
            'str': str,
            'bool': bool,
            'len': len,
        })

        # Build semaphore for any tasks that declared max_concurrency
        self._resource_semaphores: Dict[Optional[str], Semaphore] = {}
        for t in self.tasks_by_name.values():
            tag = t.resource_tag
            if tag and t.max_concurrency:
                # only one semaphore per tag
                if tag in self._resource_semaphores:
                    # pick the smallest limit if multiple tasks share the tag
                    existing = self._resource_semaphores[tag]._value
                    limit = min(existing, t.max_concurrency)
                    self._resource_semaphores[tag] = Semaphore(limit)
                else:
                    self._resource_semaphores[tag] = Semaphore(t.max_concurrency)

    def _build_graph(self):
        """
        Build:
            - self.tasks_by_name: Dict[name, TaskModel]
            - self.adj: adjacency list keyed by name
            - self.indegree: count of incoming edges keyed by name
        Also validate:
            - unique 'name' values
            - no missing dependencies
            - that each TaskModel.task exists in task_registry
        """
        self.tasks_by_name: Dict[str, TaskModel] = {
            t.name: t for t in self.pipeline.tasks
        }

        # Verify that all names are unique
        if len(self.tasks_by_name) != len(self.pipeline.tasks):
            logger.error("Duplicate task 'name' detected in pipeline.")
            raise ValueError("Duplicate task 'name' detected in pipeline.")

        # Verify that each referenced 'task' maps to a registered function
        missing_tasks: Set[str] = {
            t.task for t in self.pipeline.tasks if t.task not in task_registry
        }
        if missing_tasks:
            logger.error(f"Unknown task(s) in registry: {missing_tasks}")
            raise ValueError(f"Unknown task(s) in registry: {missing_tasks}")

        # Build adjacency & indegree maps
        self.adj: Dict[str, List[str]] = defaultdict(list)
        self.indegree: Dict[str, int] = {name: 0 for name in self.tasks_by_name}

        for t in self.tasks_by_name.values():
            for dep_name in t.depends_on:
                if dep_name not in self.tasks_by_name:
                    logger.error(
                        f"Task '{t.name}' depends on unknown task name '{dep_name}'"
                    )
                    raise ValueError(
                        f"Task '{t.name}' depends on unknown task name '{dep_name}'"
                    )
                self.adj[dep_name].append(t.name)
                self.indegree[t.name] += 1

    def _compute_layers(self) -> List[List[str]]:
        """
        Partition tasks into "layers" (batches) so that all tasks in a layer have
        indegree=0 (i.e. no unfulfilled dependencies), then remove them and repeat.
        This returns a list of lists, where each sublist is a batch of task-names
        that can run concurrently.
        """
        indegree_copy = dict(self.indegree)  # don't modify the original
        adj_copy = {u: list(v) for u, v in self.adj.items()}

        layers: List[List[str]] = []
        remaining = set(self.tasks_by_name.keys())

        while remaining:
            # Find all tasks with indegree == 0
            zero_deps = [n for n in remaining if indegree_copy[n] == 0]
            if not zero_deps:
                # cycle detected
                logger.error("Cycle detected in task dependencies")
                raise RuntimeError("Cycle detected in task dependencies")

            layers.append(zero_deps)
            # Remove these from the graph
            for u in zero_deps:
                remaining.remove(u)
                for v in adj_copy.get(u, []):
                    indegree_copy[v] -= 1

        return layers

    def _render_env(self, raw_env: Dict[str, Any]) -> Dict[str, str]:
        """
        Recursively render each value in raw_env as a Jinja2 template
        and return a flat dict of string->string to inject into os.environ.
        """
        def render_val(v: Any) -> str:
            if isinstance(v, str):
                tmpl = self._jinja_env.from_string(v)
                try:
                    return tmpl.render(**self.context)
                except jinja2.UndefinedError as e:
                    raise RuntimeError(f"Env template error in '{v}': {e}")
            else:
                # Non-string values get cast to str
                return str(v)

        return {k: render_val(v) for k, v in raw_env.items()}

    def _render_params(self, raw_params: Dict[str, Any]) -> Dict[str, Any]:
        """
        Recursively walk raw_params and render any string values as Jinja2 templates
        against self.context. For non-string or nested structures, process accordingly.
        """
        def render_value(value: Any) -> Any:
            if isinstance(value, str):
                # Treat the entire string as a Jinja2 template
                template = self._jinja_env.from_string(value)
                try:
                    return template.render(**self.context)
                except jinja2.UndefinedError as e:
                    raise RuntimeError(f"Template error in '{value}': {e}")
            elif isinstance(value, dict):
                return {k: render_value(v) for k, v in value.items()}
            elif isinstance(value, list):
                return [render_value(v) for v in value]
            else:
                # int, float, bool, etc.-leave as-is
                return value

        return render_value(raw_params)

    async def _run_single_task(self, name: str) -> None:
        """
        Execute one task by name, honoring:
         - run_if (evaluate a Jinja2 expression; skip if false)
         - retries (number of extra attempts)
         - retry_delay (seconds to sleep between attempts)
         - timeout (max seconds to wait for the underlying awaitable)
         - ignore_failure

        To unify sync vs. async:
        1) If func is a coroutine function, `coro = func(params)`
        2) Else, `coro = loop.run_in_executor(None, func, params)`
        Then we `await coro` (or `await asyncio.wait_for(coro, timeout)`).

        Records summary info (start time, attempts, status, duration, error).
        """
        task_model = self.tasks_by_name[name]
        func = task_registry[task_model.task]

        # ---- 0) SKIP-DOWNSTREAM-ON-FAILURE ----
        # If any dependency declared skip_downstream_on_failure=True and failed,
        # then skip this task.
        for dep in task_model.depends_on:
            dep_model = self.tasks_by_name[dep]
            if dep_model.skip_downstream_on_failure:
                dep_summary = self._summary.tasks.get(dep)
                if dep_summary and dep_summary.status != "success":
                    logger.info(
                        f"Task '{name}' skipped because dependency '{dep}' failed "
                        f"and skip_downstream_on_failure=True."
                    )
                    self._summary.record_skipped(name)
                    self.context[name] = None
                    return

        # ---- BRANCH GATING ----
        # If this task belongs to a branch, evaluate that branch's Jinja2 expr.
        branch_name = task_model.branch
        if branch_name:
            expr = self.pipeline.branches.get(branch_name, "")
            try:
                tmpl = self._jinja_env.from_string(expr)
                rendered = tmpl.render(**self.context).strip().lower()
            except jinja2.UndefinedError as e:
                raise RuntimeError(f"Error evaluating branch '{branch_name}': {e}")

            if rendered not in ("true", "1", "yes"):
                logger.info(f"Task '{name}' skipped because branch '{branch_name}' = '{rendered}'")
                self._summary.record_skipped(name)
                self.context[name] = None
                return

        # --- RATE-LIMITING ----
        if task_model.rate_limit is not None:
            key = task_model.rate_limit_key or task_model.name
            limiter = self._rate_limiters.get(key)
            if limiter:
                logger.debug(f"RateLimiter acquire for key={key} at rate={limiter.rate}/s")
                await limiter.acquire()

        # ---- 1) CONDITIONAL EXECUTION ----
        # A) run_unless: if provided and truthy -> skip
        if task_model.run_unless:
            tmpl_un = self._jinja_env.from_string(task_model.run_unless)
            val_un = tmpl_un.render(**self.context).strip().lower()
            if val_un in ("true", "1", "yes"):
                logger.info(f"Task '{name}' skipped because run_unless evaluated to '{val_un}'.")
                self._summary.record_skipped(name)
                self.context[name] = None
                return

        # B) run_if: if provided and falsy -> skip
        if task_model.run_if:
            try:
                tmpl_if = self._jinja_env.from_string(task_model.run_if)
                rendered = tmpl_if.render(**self.context)
            except jinja2.UndefinedError as e:
                raise RuntimeError(f"Template error in run_if for '{name}': {e}")

            # Interpret the rendered string as a boolean
            rendered_lower = rendered.strip().lower()
            if rendered_lower not in ("true", "1", "yes"):
                logger.info(f"Task '{name}' skipped because run_if evaluated to '{rendered}'.")
                # Mark skipped in summary and store None in context
                self._summary.record_skipped(name)
                self.context[name] = None

                TASK_STATUS.labels(task=name, status="skipped").inc()
                TASK_DURATION.labels(task=name, status="skipped").observe(0.0)
                return

            # else: run_if is truthy -> proceed to actual execution

        # ---- 2) PARAM RENDERING ----
        try:
            raw_params: Dict[str, Any] = task_model.params or {}
            params = self._render_params(raw_params)
        except Exception as e:
            raise RuntimeError(f"Error rendering params for task '{name}': {e}")

        max_attempts = 1 + (task_model.retries or 0)
        delay = float(task_model.retry_delay or 0.0)
        timeout = task_model.timeout  # None or float
        ignore_failure = bool(task_model.ignore_failure)

        # Prepare execution as an inner coroutine (to allow semaphore)
        async def execute():
            # ---- ENVIRONMENT INJECTION ----
            raw_env = task_model.env or {}
            if raw_env:
                try:
                    env_vars = self._render_env(raw_env)
                except Exception as e:
                    raise RuntimeError(f"Error rendering env for task '{name}': {e}")

                # Save current values
                _old = {k: os.environ.get(k) for k in env_vars}
                # Inject new ones
                os.environ.update(env_vars)
            else:
                env_vars = {}
                _old = {}

            self._summary.record_start(name)

            attempt = 0

            while True:
                attempt += 1
                # Always run in executor so resource limits apply
                loop = asyncio.get_running_loop()
                coro = loop.run_in_executor(
                    None,
                    limit_and_call,
                    func,
                    params,
                    task_model.cpu_time,
                    task_model.memory,
                )
                # if asyncio.iscoroutinefunction(func):
                #     # async functions can't have resource limits easily; run in a thread
                #     def _sync_wrapper(p):
                #         return limit_and_call(func, p)
                #     coro = func(params)     # async tasks: relay on timeout only
                # else:
                #     # Offload sync function to a threadpool so it doesn't block the loop
                #     loop = asyncio.get_running_loop()
                #     # wrap sync function so it sets resource limits before calling
                #     coro = loop.run_in_executor(
                #         None,
                #         limit_and_call,
                #         func,
                #         params
                #     )

                try:
                    # capture whatever the task returned
                    start = time.time()
                    if timeout and timeout > 0:
                        result = await asyncio.wait_for(coro, timeout=timeout)
                    else:
                        result = await coro
                    dur = time.time() - start

                    # record metrics
                    TASK_STATUS.labels(
                        pipeline=self.pipeline_name,
                        task=name,
                        status="success"
                    ).inc()
                    TASK_DURATION.labels(
                        pipeline=self.pipeline_name,
                        task=name,
                        status="success"
                    ).observe(dur)

                    # Record success and store result in context
                    logger.info(f"Task '{name}' succeeded on attempt {attempt}/{max_attempts}")
                    self._summary.record_success(name, attempt)

                    # 📦 unpack dict‐returns into context, or bind single value
                    if isinstance(result, dict):
                        for k, v in result.items():
                            if k in self.context:
                                logger.warning(f"Context key {k!r} overwritten by task '{name}'")
                            self.context[k] = v
                    else:
                        self.context[name] = result

                    return
                except asyncio.TimeoutError as te:
                    # Timeout on this attempt
                    if attempt >= max_attempts:
                        msg = f"Task '{name}' timed out after {timeout}s (attempt {attempt}/{max_attempts})"
                        if ignore_failure:
                            logger.error(msg + " — but ignore_failure=True, continuing.")
                            self._summary.record_failed_ignored(name, attempt, te)
                            # record metrics
                            TASK_STATUS.labels(
                                pipeline=self.pipeline_name,
                                task=name,
                                status="failed_ignored"
                            ).inc()
                            TASK_DURATION.labels(
                                pipeline=self.pipeline_name,
                                task=name,
                                status="failed_ignored"
                            ).observe(time.time() - self._summary.tasks[name].start_time)
                            return
                        else:
                            logger.error(msg)
                            self._summary.record_failed_abort(name, attempt, te)
                            TASK_STATUS.labels(
                                pipeline=self.pipeline_name,
                                task=name,
                                status="failed_abort"
                            ).inc()
                            TASK_DURATION.labels(
                                pipeline=self.pipeline_name,
                                task=name,
                                status="failed_abort"
                            ).observe(time.time() - self._summary.tasks[name].start_time)
                            raise RuntimeError(msg)
                    else:
                        # Log a warning and sleep before next attempt
                        logger.warning(
                            f"⏱️ Task '{name}' timed out after {timeout}s "
                            f"(attempt {attempt}/{max_attempts}). Retrying in {delay:.1f}s..."
                        )
                        if delay > 0:
                            await asyncio.sleep(delay)

                except Exception as exc:
                    # Real exception from the task body
                    if attempt >= max_attempts:
                        msg = f"Task '{name}' (func={task_model.task}) failed permanently with: {exc!r}"
                        if task_model.ignore_failure:
                            logger.error(msg + " — but ignore_failure=True, continuing.")
                            self._summary.record_failed_ignored(name, attempt, exc)
                            TASK_STATUS.labels(
                                pipeline=self.pipeline_name,
                                task=name,
                                status="failed_ignored"
                            ).inc()
                            TASK_DURATION.labels(
                                pipeline=self.pipeline_name,
                                task=name,
                                status="failed_ignored"
                            ).observe(time.time() - self._summary.tasks[name].start_time)
                            # Even though failure is ignored, we set context[name] = None
                            self.context[name] = None
                            return
                        else:
                            logger.error(msg)
                            self._summary.record_failed_abort(name, attempt, exc)
                            TASK_STATUS.labels(
                                pipeline=self.pipeline_name,
                                task=name,
                                status="failed_abort"
                            ).inc()
                            TASK_DURATION.labels(
                                pipeline=self.pipeline_name,
                                task=name,
                                status="failed_abort"
                            ).observe(time.time() - self._summary.tasks[name].start_time)
                            raise
                    else:
                        logger.warning(
                            f"⚠️ Task '{name}' (func={task_model.task}) failed with {exc!r} "
                            f"(attempt {attempt}/{max_attempts}). Retrying in {delay:.1f}s..."
                        )
                        if delay > 0:
                            await asyncio.sleep(delay)

                finally:
                    # Restore environment
                    if env_vars:
                        for k, old_val in _old.items():
                            if old_val is None:
                                os.environ.pop(k, None)
                            else:
                                os.environ[k] = old_val

        # Acquire semaphore if resource_tag is set
        sem = None
        tag = task_model.resource_tag
        if tag and tag in self._resource_semaphores:
            sem = self._resource_semaphores[tag]

        if sem:
            async with sem:
                await execute()
        else:
            await execute()

    async def _run_layer(self, names: List[str]) -> None:
        """
        Given a batch of task-names, run them concurrently using asyncio.gather.
        """
        coros = [self._run_single_task(n) for n in names]
        await asyncio.gather(*coros)

    def _topo_sort(self) -> List[str]:
        """
        Kahn's algorithm on self.indegree & self.adj to produce
        a topologically sorted list of 'name' keys. Detects cycles.
        """
        queue = deque([n for n, deg in self.indegree.items() if deg == 0])
        order: List[str] = []

        while queue:
            u = queue.popleft()
            order.append(u)
            for v in self.adj[u]:
                self.indegree[v] -= 1
                if self.indegree[v] == 0:
                    queue.append(v)

        if len(order) != len(self.tasks_by_name):
            raise RuntimeError("Cycle detected in task dependencies")

        return order

    def run(self) -> PipelineRunSummary:
        """
        1. Load all plugins → ensure task_registry is populated
        2. Compute dependency “layers”
        3. For each layer, run all tasks concurrently (with retry logic baked in)
        """
        load_plugins()

        # Re-validate that every TaskModel.task is in registry
        missing = [t.task for t in self.tasks_by_name.values() if t.task not in task_registry]
        if missing:
            logger.error(f"Missing registered functions: {missing}")
            raise RuntimeError(f"Missing registered functions: {missing}")

        logger.info("Starting pipeline execution...")
        layers = self._compute_layers()
        # Run each layer in sequence, but tasks within each layer in parallel
        for layer in layers:
            logger.info(f"Executing layer: {layer}")
            asyncio.run(self._run_layer(layer))

        # Return the summary for further handling (e.g., JSON export)
        return self._summary

    def print_dag(self) -> None:
        """
        ASCII view of each task name and its dependencies.
        """
        for name, t in self.tasks_by_name.items():
            deps = ", ".join(t.depends_on) if t.depends_on else "—"
            logger.info(f"{name:20} depends on → {deps}")

    def to_dot(self) -> str:
        """
        Generate a Graphviz DOT representation of the DAG.
        Each task is a node labeled "<name>\n<task_func>".
        Edges go from each dependency to the dependent.
        """
        lines: List[str] = [
            "digraph NovaPipe {",
            "    rankdir=LR;",  # left-to-right orientation
            "    node [shape=box, style=filled, fillcolor=lightgray];",
            ""
        ]

        # Declare nodes (with label = name + function)
        for name, t in self.tasks_by_name.items():
            label = f"{name}\\n({t.task})"
            lines.append(f'    "{name}" [label="{label}"];')

        lines.append("")  # blank line before edges

        # Declare edges
        for name, t in self.tasks_by_name.items():
            for dep_name in t.depends_on:
                lines.append(f'    "{dep_name}" -> "{name}";')

        lines.append("}")
        return "\n".join(lines)

print_dag()

ASCII view of each task name and its dependencies.

Source code in novapipe/runner.py

def print_dag(self) -> None:
    """
    ASCII view of each task name and its dependencies.
    """
    for name, t in self.tasks_by_name.items():
        deps = ", ".join(t.depends_on) if t.depends_on else "—"
        logger.info(f"{name:20} depends on → {deps}")

run()

1. Load all plugins → ensure task_registry is populated
2. Compute dependency “layers”
3. For each layer, run all tasks concurrently (with retry logic baked in)

Source code in novapipe/runner.py

def run(self) -> PipelineRunSummary:
    """
    1. Load all plugins → ensure task_registry is populated
    2. Compute dependency “layers”
    3. For each layer, run all tasks concurrently (with retry logic baked in)
    """
    load_plugins()

    # Re-validate that every TaskModel.task is in registry
    missing = [t.task for t in self.tasks_by_name.values() if t.task not in task_registry]
    if missing:
        logger.error(f"Missing registered functions: {missing}")
        raise RuntimeError(f"Missing registered functions: {missing}")

    logger.info("Starting pipeline execution...")
    layers = self._compute_layers()
    # Run each layer in sequence, but tasks within each layer in parallel
    for layer in layers:
        logger.info(f"Executing layer: {layer}")
        asyncio.run(self._run_layer(layer))

    # Return the summary for further handling (e.g., JSON export)
    return self._summary

to_dot()

    Generate a Graphviz DOT representation of the DAG.
    Each task is a node labeled "<name>

". Edges go from each dependency to the dependent.

Source code in novapipe/runner.py

def to_dot(self) -> str:
    """
    Generate a Graphviz DOT representation of the DAG.
    Each task is a node labeled "<name>\n<task_func>".
    Edges go from each dependency to the dependent.
    """
    lines: List[str] = [
        "digraph NovaPipe {",
        "    rankdir=LR;",  # left-to-right orientation
        "    node [shape=box, style=filled, fillcolor=lightgray];",
        ""
    ]

    # Declare nodes (with label = name + function)
    for name, t in self.tasks_by_name.items():
        label = f"{name}\\n({t.task})"
        lines.append(f'    "{name}" [label="{label}"];')

    lines.append("")  # blank line before edges

    # Declare edges
    for name, t in self.tasks_by_name.items():
        for dep_name in t.depends_on:
            lines.append(f'    "{dep_name}" -> "{name}";')

    lines.append("}")
    return "\n".join(lines)

RateLimiter

Simple sliding-window rate limiter: up to rate calls per per seconds.

Source code in novapipe/runner.py

class RateLimiter:
    """
    Simple sliding-window rate limiter: up to `rate` calls per `per` seconds.
    """
    def __init__(self, rate: float, per: float = 1.0):
        self.rate = rate
        self.per = per
        self.calls: Deque[float] = deque()

    async def acquire(self):
        now = asyncio.get_running_loop().time()
        # Remove timestamps older than window
        while self.calls and self.calls[0] <= now - self.per:
            self.calls.popleft()

        if len(self.calls) < self.rate:
            # under limit: record and proceed
            self.calls.append(now)
            return
        # at capacity: compute next allowed time
        earliest = self.calls[0]
        wait = earliest + self.per - now
        await asyncio.sleep(wait)
        # after waiting, slide window and record
        now2 = asyncio.get_running_loop().time()
        self.calls.popleft()
        self.calls.append(now2)

TaskMetrics

Stores summary info for one task

• attempts: total attempts made (1 + retries)
• status: "success", "failed_ignored", or "failed_abort"
• duration_secs: wall‐clock time from first attempt start to final outcome
• error: error message (if any; null on success)

Source code in novapipe/runner.py

class TaskMetrics:
    """
    Stores summary info for one task:
      - attempts: total attempts made (1 + retries)
      - status: "success", "failed_ignored", or "failed_abort"
      - duration_secs: wall‐clock time from first attempt start to final outcome
      - error: error message (if any; null on success)
    """
    def __init__(self, name: str):
        self.name: str = name
        self.attempts: int = 0
        self.status: Optional[str] = None
        self.start_time: Optional[float] = None
        self.duration_secs: Optional[float] = None
        self.error: Optional[str] = None

    def to_dict(self) -> Dict[str, Any]:
        return {
            "name": self.name,
            "attempts": self.attempts,
            "status": self.status,
            "duration_secs": self.duration_secs,
            "error": self.error,
        }

limit_and_call(fn, params, cpu_time=None, memory=None)

Apply RLIMIT_CPU and RLIMIT_AS (if given), then call fn(params). If fn(params) returns a coroutine, run it via asyncio.run(). Return the final result.

Source code in novapipe/runner.py

def limit_and_call(fn, params, cpu_time=None, memory=None):
    """
    Apply RLIMIT_CPU and RLIMIT_AS (if given), then call fn(params).
    If fn(params) returns a coroutine, run it via asyncio.run().
    Return the final result.
    """
    if _HAS_RESOURCE:
        # Apply CPU limit
        if cpu_time is not None:
            resource.setrlimit(resource.RLIMIT_CPU, (cpu_time, cpu_time))
        # Apply memory limit (address space)
        if memory is not None:
            resource.setrlimit(resource.RLIMIT_AS, (memory, memory))
    else:
        # Windows or missing resource module: we can't enforce limits
        if cpu_time is not None or memory is not None:
            # only log once per call
            logger.warning(
                "Resource limits requested but not supported on this platform—"
                "skipping cpu_time=%r, memory=%r", cpu_time, memory
            )

    result = fn(params)
    if asyncio.iscoroutine(result):
        # run any coroutine to completion
        return asyncio.run(result)
    return result

---

PipelineRunSummary (in novapipe.runner)

Collects a dict of TaskSummary, keyed by task name.

Source code in novapipe/runner.py

class PipelineRunSummary:
    """
    Collects a dict of TaskSummary, keyed by task name.
    """
    def __init__(self):
        self.tasks: Dict[str, TaskMetrics] = {}

    def record_start(self, name: str):
        ts = TaskMetrics(name)
        self.tasks[name] = ts
        ts.start_time = time.time()  # for intermediate tracking

    def record_success(self, name: str, attempts: int):
        ts = self.tasks[name]
        ts.attempts = attempts
        ts.status = "success"
        ts.duration_secs = time.time() - ts.start_time
        ts.error = None

    def record_failed_ignored(self, name: str, attempts: int, error: Exception):
        ts = self.tasks[name]
        ts.attempts = attempts
        ts.status = "failed_ignored"
        ts.duration_secs = time.time() - ts.start_time
        ts.error = repr(error)

    def record_failed_abort(self, name: str, attempts: int, error: Exception):
        ts = self.tasks[name]
        ts.attempts = attempts
        ts.status = "failed_abort"
        ts.duration_secs = time.time() - ts.start_time
        ts.error = repr(error)

    def record_skipped(self, name: str):
        ts = TaskMetrics(name)
        ts.attempts = 0
        ts.status = "skipped"
        ts.duration_secs = 0.0
        ts.error = None
        ts.start_time = time.time()
        self.tasks[name] = ts

    def to_list(self) -> List[Dict[str, Any]]:
        return [ts.to_dict() for ts in self.tasks.values()]

---

Helper Utilities

• limit_and_call: Apply resource limits before executing tasks.
• set_plugin_pins: Pin plugin versions for conflict resolution.

Refer to the sections above for these utilities where applicable.

---

Automatically generated.sections will be populated by mkdocstrings when building the docs.