require

execution dependencies · maintainer machinable-org · upstream

Gate a run on its dependencies being cached.

Install

Declare the remote in your project's provider; the module is fetched on first use:

# interface/project.py — your project's provider
from machinable import Project


class MyProject(Project):
    def on_resolve_remotes(self):
        return {
            "require": "url+https://raw.githubusercontent.com/machinable-org/machinable/5c73557505915231582d680ae26c6404b747e812/integrations/require/require.py",
        }

The module is fetched once and cached in interface/remotes/; machinable fetch require downloads it without importing so you can inspect the code first. See Resolving remotes for how pulling, pinning, and updating work.

Asserts that components have been computed instead of running them, which is useful to fail fast when a pipeline stage expects cached inputs.

Usage

from machinable import get

with get("require"):
    ...  # components to check

Configuration

Option	Type	Default
prevent_commit	bool	True

Source

integrations/require/require.py

require.py

from pydantic import BaseModel

from machinable import Execution


class Require(Execution):
    class Config(BaseModel):
        prevent_commit: bool = True

    def add(
        self,
        executable,
    ):
        if not self.config.prevent_commit:
            return super().add(executable)

        if isinstance(executable, list | tuple):
            for _executable in executable:
                self.add(_executable)
            return self

        if not executable.cached():
            raise RuntimeError(
                f"Execution required for {executable.module} <{executable.id}>"
            )

        return super().add(executable)

    def on_before_dispatch(self):
        raise RuntimeError(
            "Execution is required:\n- "
            + "\n- ".join(
                self.pending_executables.map(lambda x: x.module + " <" + x.id + ">")
            )
        )