API Reference

sdag.

commands

CLI commands.

Commands executed through the CLI. It's the point of contact between the Python API and the Rust core.

compile_and_return_dag(path, extra_metadata, input_kwargs)

Compile a pipeline.

Parameters:

Name	Type	Description	Default
path	str	Pipeline name or import string.	required
extra_metadata	Any	Extra metadata that will be attached to the pipeline.	required
input_kwargs	dict[str, Any]	Pipeline input kwargs.	required

Returns:

Name	Type	Description
DAG	DAG	Compiled graph.

Source code in python/sdag/commands.py

def compile_and_return_dag(
    path: str, extra_metadata: Any, input_kwargs: dict[str, Any]
) -> DAG:
    """Compile a pipeline.

    Args:
        path (str): Pipeline name or import string.
        extra_metadata (Any): Extra metadata that will
            be attached to the pipeline.
        input_kwargs (dict[str, Any]): Pipeline input kwargs.

    Returns:
        DAG: Compiled graph.
    """
    logger.info("Compiling pipeline '%s'", path)
    pipeline = find_pipeline_by_name(path)
    dag = pipeline.compile(path, **input_kwargs)
    dag.meta.code_hash = pipeline.compute_code_hash()
    dag.meta.kwargs |= input_kwargs
    if extra_metadata is not None:
        dag.meta.extra = json.loads(extra_metadata)

    cfg = get_pyproj()
    apply_pyproj_configs(dag, cfg)

    logger.info("Compiled pipeline '%s'", path)

    return dag

compile_pipeline(args, extras)

Compile a pipeline.

Parameters:

Name	Type	Description	Default
args	Namespace	Parsed args.	required
extras	dict[str, Any]	Extra args turned into input kwargs.	required

Returns:

Name	Type	Description
Path	Path	Path where the compiled JSON has been saved.

Source code in python/sdag/commands.py

def compile_pipeline(args: Namespace, extras: dict[str, Any]) -> Path:
    """Compile a pipeline.

    Args:
        args (Namespace): Parsed args.
        extras (dict[str, Any]): Extra args turned into input kwargs.

    Returns:
        Path: Path where the compiled JSON has been saved.
    """
    pyproj = get_pyproj()
    pyproj.join_cli_args(args)
    configure_logging(pyproj.log_level)

    dag = compile_and_return_dag(
        path=args.pipeline,
        extra_metadata=args.extra_metadata,
        input_kwargs=extras,
    )

    dag_json = dag.model_dump_json(indent=4, warnings="none", by_alias=True)
    dst_path = Path(pyproj.compiled_dag_dir)
    dst_path.mkdir(parents=True, exist_ok=True)
    dag_name = dag.meta.pipeline_name
    dst_name = args.name if args.name is not None else f"{dag_name}.json"

    dst_path = dst_path / dst_name
    logger.info("Saving pipeline to '%s'", dst_path)

    with dst_path.open("w") as f:
        f.write(dag_json)

    return dst_path

describe_pipeline(args, extras)

Describe a pipeline arguments, caching, and artifacts.

Parameters:

Name	Type	Description	Default
args	Namespace	Parsed args.	required
extras	dict[str, Any]	Extra arguments used for compiling.	required

Source code in python/sdag/commands.py

def describe_pipeline(args: Namespace, extras: dict[str, Any]) -> None:
    """Describe a pipeline arguments, caching, and artifacts.

    Args:
        args (Namespace): Parsed args.
        extras (dict[str, Any]): Extra arguments used for compiling.
    """
    pyproj = get_pyproj()
    pyproj.join_cli_args(args)
    configure_logging(pyproj.log_level)

    if args.pipeline.endswith(".json"):
        path = _find_compiled_path(args.pipeline)
    else:
        path = compile_pipeline(args, extras)

    core.describe_pipeline(str(path), log_level=pyproj.log_level)

get_task(name, pipeline_name)

Import a task.

Parameters:

Name	Type	Description	Default
name	str	Task name.	required
pipeline_name	str	Name of the pipeline using the task.	required

Raises:

Type	Description
DAGNotFoundError	The task cannot be found.

Returns:

Name	Type	Description
Task	Task	Task.

Source code in python/sdag/commands.py

def get_task(name: str, pipeline_name: str) -> Task:
    """Import a task.

    Args:
        name (str): Task name.
        pipeline_name (str): Name of the pipeline using the task.

    Raises:
        DAGNotFoundError: The task cannot be found.

    Returns:
        Task: Task.
    """
    pipeline = find_pipeline_by_name(pipeline_name)
    if name in pipeline.tasks:
        return pipeline.tasks[name]
    if name in master.tasks:
        return master.tasks[name]

    msg = f"Task '{name}' of pipeline '{pipeline_name}' not found"
    raise DAGNotFoundError(msg)

kill_pipeline(args, extras)

Kill a pipeline.

Parameters:

Name	Type	Description	Default
args	Namespace	Parsed args.	required
extras	dict[str, Any]	Extra args (not allowed).	required

Raises:

Type	Description
ExtraCLIArgsError	Extra argument used.

Source code in python/sdag/commands.py

def kill_pipeline(args: Namespace, extras: dict[str, Any]) -> None:
    """Kill a pipeline.

    Args:
        args (Namespace): Parsed args.
        extras (dict[str, Any]): Extra args (not allowed).

    Raises:
        ExtraCLIArgsError: Extra argument used.
    """
    if extras:
        raise ExtraCLIArgsError(extras)

    pyproj = get_pyproj()
    pyproj.join_cli_args(args)
    configure_logging(pyproj.log_level)

    pipeline_name = args.pipeline
    if args.pipeline.endswith(".json"):
        dag = _parse_compiled_pipeline(args.pipeline)
        pipeline_name = dag.meta.pipeline_name

    core.kill_run(
        pipeline_name=pipeline_name,
        pipeline_hash=args.hash,
        log_level=pyproj.log_level,
    )

list_pipelines(args, extras)

List pipelines and runs.

By default, pipelines are listed. If the pipeline name is specified, then run hashes are listed.

Parameters:

Name	Type	Description	Default
args	Namespace	Parsed args.	required
extras	dict[str, Any]	Extra arguments (not allowed)	required

Raises:

Type	Description
ExtraCLIArgsError	Extra argument used.

Source code in python/sdag/commands.py

def list_pipelines(args: Namespace, extras: dict[str, Any]) -> None:
    """List pipelines and runs.

    By default, pipelines are listed. If the pipeline name is
    specified, then run hashes are listed.

    Args:
        args (Namespace): Parsed args.
        extras (dict[str, Any]): Extra arguments (not allowed)

    Raises:
        ExtraCLIArgsError: Extra argument used.
    """
    if extras:
        raise ExtraCLIArgsError(extras)

    pyproj = get_pyproj()
    pyproj.join_cli_args(args)
    configure_logging(pyproj.log_level)

    if args.pipeline is not None:
        core.print_runs(
            pipeline_name=args.pipeline, log_level=pyproj.log_level
        )
        return

    dags = find_all_pipelines()
    sorted_dags = sorted(dags.items(), key=lambda kv: kv[0])
    names = [name for name, _ in sorted_dags]
    paths = [path for _, path in sorted_dags]

    core.print_pipeline_list(names, paths)

print_status(args, extras)

Describe a pipeline arguments, caching, and artifacts.

Parameters:

Name	Type	Description	Default
args	Namespace	Parsed args.	required
extras	dict[str, Any]	Extra arguments used for compiling.	required

Source code in python/sdag/commands.py

def print_status(args: Namespace, extras: dict[str, Any]) -> None:
    """Describe a pipeline arguments, caching, and artifacts.

    Args:
        args (Namespace): Parsed args.
        extras (dict[str, Any]): Extra arguments used for compiling.
    """
    if extras:
        raise ExtraCLIArgsError(extras)

    pyproj = get_pyproj()
    pyproj.join_cli_args(args)
    configure_logging(pyproj.log_level)
    core.print_status(
        args.pipeline, pipeline_hash=args.hash, log_level=pyproj.log_level
    )

prune_cache(args, extras)

Prune the cache of tasks or pipelines.

Parameters:

Name	Type	Description	Default
args	Namespace	Parsed args.	required
extras	dict[str, Any]	Extra args (not allowed).	required

Raises:

Type	Description
ExtraCLIArgsError	Extra argument used.

Source code in python/sdag/commands.py

def prune_cache(args: Namespace, extras: dict[str, Any]) -> None:
    """Prune the cache of tasks or pipelines.

    Args:
        args (Namespace): Parsed args.
        extras (dict[str, Any]): Extra args (not allowed).

    Raises:
        ExtraCLIArgsError: Extra argument used.
    """
    if args.pipeline is None and not args.task:
        raise ExtraCLIArgsError(extras)

    pyproj = get_pyproj()
    pyproj.join_cli_args(args)
    configure_logging(pyproj.log_level)

    if args.pipeline is not None and not args.task:
        dag = _get_dag_from_name_import_or_json(
            path=args.pipeline,
            extra_metadata=args.extra_metadata,
            input_kwargs=extras,
        )

        tasks = _find_cacheable_tasks(dag)
        for task in tasks:
            pipeline_name = task.pipeline if task.scope == "local" else None
            core.prune_cache(
                task_name=task.name,
                pipeline_name=pipeline_name,
                allow_full_prune=False,
                log_level=pyproj.log_level,
            )

    else:
        name = args.pipeline
        if name is not None and ":" in name:
            name = name.split(":")[-1]

        for task in args.task:
            core.prune_cache(
                task_name=task,
                pipeline_name=name,
                allow_full_prune=True,
                log_level=pyproj.log_level,
            )

restart_run(args, extras)

Restart a run.

Parameters:

Name	Type	Description	Default
args	Namespace	Parsed args.	required
extras	dict[str, Any]	Extra args (not allowed).	required

Raises:

Type	Description
ExtraCLIArgsError	Extra argument used.

Source code in python/sdag/commands.py

def restart_run(args: Namespace, extras: dict[str, Any]) -> None:
    """Restart a run.

    Args:
        args (Namespace): Parsed args.
        extras (dict[str, Any]): Extra args (not allowed).

    Raises:
        ExtraCLIArgsError: Extra argument used.
    """
    if extras:
        raise ExtraCLIArgsError(extras)

    pyproj = get_pyproj()
    pyproj.join_cli_args(args)
    configure_logging(pyproj.log_level)

    pipeline_name = args.pipeline
    if args.pipeline.endswith(".json"):
        dag = _parse_compiled_pipeline(args.pipeline)
        pipeline_name = dag.meta.pipeline_name

    core.restart_run(
        pipeline_name=pipeline_name,
        pipeline_hash=args.hash,
        max_concurrency=pyproj.max_concurrency,
        time_between_polls=pyproj.time_between_polls,
        log_level=pyproj.log_level,
        retry=False,
        fail_fast=pyproj.fail_fast,
    )

retry_run(args, extras)

Retry a run.

Parameters:

Name	Type	Description	Default
args	Namespace	Parsed args.	required
extras	dict[str, Any]	Extra args (not allowed).	required

Raises:

Type	Description
ExtraCLIArgsError	Extra argument used.

Source code in python/sdag/commands.py

def retry_run(args: Namespace, extras: dict[str, Any]) -> None:
    """Retry a run.

    Args:
        args (Namespace): Parsed args.
        extras (dict[str, Any]): Extra args (not allowed).

    Raises:
        ExtraCLIArgsError: Extra argument used.
    """
    if extras:
        raise ExtraCLIArgsError(extras)

    pyproj = get_pyproj()
    pyproj.join_cli_args(args)
    configure_logging(pyproj.log_level)

    pipeline_name = args.pipeline
    if args.pipeline.endswith(".json"):
        dag = _parse_compiled_pipeline(args.pipeline)
        pipeline_name = dag.meta.pipeline_name

    core.restart_run(
        pipeline_name=pipeline_name,
        pipeline_hash=args.hash,
        max_concurrency=pyproj.max_concurrency,
        time_between_polls=pyproj.time_between_polls,
        log_level=pyproj.log_level,
        retry=True,
        fail_fast=pyproj.fail_fast,
    )

run_pipeline(args, extras)

Run a pipeline.

Parameters:

Name	Type	Description	Default
args	Namespace	Parsed arguments.	required
extras	dict[str, Any]	Extra arguments only used if the pipeline must be compiled.	required

Source code in python/sdag/commands.py

def run_pipeline(args: Namespace, extras: dict[str, Any]) -> None:
    """Run a pipeline.

    Args:
        args (Namespace): Parsed arguments.
        extras (dict[str, Any]): Extra arguments only used
            if the pipeline must be compiled.
    """
    pyproj = get_pyproj()
    pyproj.join_cli_args(args)
    configure_logging(pyproj.log_level)

    if args.pipeline.endswith(".json"):
        path = _find_compiled_path(args.pipeline)
    else:
        path = compile_pipeline(args, extras)

    core.run(
        pipeline_path=str(path),
        max_concurrency=pyproj.max_concurrency,
        time_between_polls=pyproj.time_between_polls,
        log_level=pyproj.log_level,
        slurm_grace_period=pyproj.slurm_grace_period,
        max_concurrent_runs=pyproj.max_concurrent_runs,
        fail_fast=pyproj.fail_fast,
    )

run_task(args, extras)

Run a single task without starting the scheduler.

Parameters:

Name	Type	Description	Default
args	Namespace	Parsed args.	required
extras	dict[str, Any]	Extra arguments used for compilation.	required

Source code in python/sdag/commands.py

def run_task(args: Namespace, extras: dict[str, Any]) -> None:
    """Run a single task without starting the scheduler.

    Args:
        args (Namespace): Parsed args.
        extras (dict[str, Any]): Extra arguments used for compilation.
    """
    pyproj = get_pyproj()
    pyproj.join_cli_args(args)
    configure_logging(pyproj.log_level)

    task = get_task(name=args.task, pipeline_name=args.pipeline)
    task = TaskNode(
        uid=0,
        root_uid=-1,
        name=task.name,
        pipeline_name=args.pipeline,
        fn_name=task.fn.__name__,
        cache=task.cache,
        code_hash=task.code_hash,
        mode=task.mode,
        cmd=task.cmd,
        scope=task.scope,
        retries=0,
        script=task.script,
        kwargs=[Kwarg(key=k, value=v) for k, v in extras.items()],
    )

    pyproj = get_pyproj()
    apply_pyproj_to_task(task, pyproj)
    task_serialized = task.model_dump_json(warnings="none", by_alias=True)
    core.run_single_task(
        task_serialized,
        log_level=pyproj.log_level,
        time_between_polls=pyproj.time_between_polls,
        slurm_grace_period=pyproj.slurm_grace_period,
        max_concurrent_runs=pyproj.max_concurrent_runs,
    )

view_pipeline(args, extras)

Visualize a pipeline in the terminal.

Parameters:

Name	Type	Description	Default
args	Namespace	Parsed args.	required
extras	dict[str, Any]	Extra arguments only used if the pipeline is not read from JSON.	required

Source code in python/sdag/commands.py

def view_pipeline(args: Namespace, extras: dict[str, Any]) -> None:
    """Visualize a pipeline in the terminal.

    Args:
        args (Namespace): Parsed args.
        extras (dict[str, Any]): Extra arguments only used
            if the pipeline is not read from JSON.
    """
    from sdag.visualization import MermaidGenerator

    pyproj = get_pyproj()
    pyproj.join_cli_args(args)
    configure_logging(pyproj.log_level)

    dag = _get_dag_from_name_import_or_json(
        path=args.pipeline,
        extra_metadata=args.extra_metadata,
        input_kwargs=extras,
    )

    mermaid_gen = MermaidGenerator(squeeze=args.squeeze)
    mermaid = mermaid_gen.generate_mermaid_string(dag)
    core.view_pipeline(mermaid=mermaid, log_level=pyproj.log_level)

compiler

Compiler.

It relies on two global objects everyone has access to, so it is not thread safe.

compiler = DAGCompiler() module-attribute

Compiler.

master = SDAG() module-attribute

Keeps track of imported pipelines and tasks.

DAGCompiler

Compiler.

It does most of the magic, keeps track of the graph and updates it when tasks are executed.

Attributes:

Name	Type	Description
outer	list[CompiledDAG]	Pipelines being compiled.
active	CompiledDAG | None	Active pipeline compiled.
_uid	int	Used to assign monotonically increasing uids.

Source code in python/sdag/compiler.py

class DAGCompiler:
    """Compiler.

    It does most of the magic, keeps track of
    the graph and updates it when tasks are
    executed.

    Attributes:
        outer (list[CompiledDAG]): Pipelines being compiled.
        active (CompiledDAG | None): Active pipeline compiled.
        _uid (int): Used to assign monotonically increasing uids.
    """

    def __init__(self):
        """Initialize the compiler."""
        self.outer: list[CompiledDAG] = []
        self.active: CompiledDAG | None = None
        self._uid = 0

    def reset(self) -> None:
        """Reset the compiler state.

        Called before every compiled pipeline.
        """
        self.outer = []
        self.active = None
        self._uid = 0

    def get_uid(self) -> int:
        """Get a new uid.

        Returns:
            int: Unique id.
        """
        uid = self._uid
        self._uid += 1
        return uid

    def register(self, node: NodeUnion) -> None:
        """Register a node in the graph.

        If a branch scope is currently active, the node is given a
        branch edge to it, tying it to that branch's true/false path.

        Args:
            node (NodeUnion): Node.

        Raises:
            DAGNotSetError: Attempting to register a node
                but no pipelines are being compiled.
        """
        if self.active is None:
            raise DAGNotSetError

        if not node.pipeline_name:
            name = self.active.dag.meta.pipeline_name
            node.pipeline_name = name
            node.root_uid = self.active.root.uid

        self.active.dag.nodes.append(node)

        anchor = self.active.scope.anchor
        if anchor is not None:
            node.add_branch_edge(anchor[0], anchor[1])

    @property
    def current_scope(self) -> TraceScope:
        """TraceScope currently active in the pipeline being traced.

        Raises:
            DAGNotSetError: No pipelines are being compiled.

        Returns:
            TraceScope: Current scope.
        """
        if self.active is None:
            raise DAGNotSetError
        return self.active.scope

    def push_scope(self, anchor: tuple[int, bool] | None) -> TraceScope:
        """Enter a new lexical scope, e.g. an If/Elif/Else body.

        Args:
            anchor (tuple[int, bool] | None): (branch_uid, branch_value)
                stamped on every node registered in the new scope.

        Returns:
            TraceScope: The newly active scope.
        """
        scope = TraceScope(parent=self.current_scope, anchor=anchor)
        self.active.scope = scope  # type: ignore
        return scope

    def pop_scope(self) -> TraceScope:
        """Exit the current lexical scope, restoring its parent.

        Raises:
            DAGNotSetError: No pipelines are being compiled, or
                attempting to pop the pipeline's root scope.

        Returns:
            TraceScope: The scope that was just closed.
        """
        closed = self.current_scope
        if closed.parent is None:
            raise DAGNotSetError
        self.active.scope = closed.parent  # type: ignore
        return closed

    def close_clause(
        self,
        branch_uid: int,
        next_edge_value: bool,
        terminal: bool = False,  # noqa: FBT002
    ) -> None:
        """Record a closed branch for the next Elif/Else.

        Args:
            branch_uid (int): Uid of the BranchNode the clause evaluated.
            next_edge_value (bool): Branch edge value a following
                Elif/Else must depend on.
            terminal (bool): True if this clause was an Else, closing
                the chain to further Elif/Else.
        """
        self.current_scope.last_clause = ClosedClause(
            branch_uid=branch_uid,
            next_edge_value=next_edge_value,
            terminal=terminal,
        )

    def set_dag(self, root: RootNode, import_path: str | None = None) -> None:
        """Set a new dag.

        Every time a new pipeline is called.

        Args:
            root (RootNode): Pipeline root node.
            import_path (str | None, optional): Pipeline import path.
                If null, it defaults to the pipeline name.
                Defaults to None.
        """
        name = root.pipeline_name
        import_path = import_path if import_path is not None else name
        dag = DAG(meta=DAGMeta(pipeline_name=name, import_path=import_path))
        end = EndNode(
            uid=self.get_uid(), root_uid=root.uid, pipeline_name=name
        )
        active = CompiledDAG(dag=dag, root=root, end=end)

        if self.active is not None:
            self.outer.append(self.active)

        self.active = active
        self.register(root)

    def get_dag(self) -> tuple[DAG, RootNode, EndNode]:
        """Detach a dag from the compiler.

        Raises:
            DAGNotSetError: No DAGs are set.

        Returns:
            tuple[DAG, RootNode, EndNode]: Graph, root, and endnode.
        """
        if self.active is None:
            raise DAGNotSetError

        dag = self.active.dag
        root = self.active.root
        end = self.active.end
        parent_uids = set()

        for node in dag.nodes:
            if not node.parents and node.uid not in (root.uid, end.uid):
                node.add_logical_edge(root.uid)
            for parent in node.parents:
                parent_uids.add(parent.uid)

        for node in dag.nodes:
            if node.uid not in parent_uids:
                end.add_logical_edge(node.uid)
        self.register(end)

        if self.outer:
            self.active = self.outer.pop()
            self.active.dag.nodes.extend(dag.nodes)

            anchor = self.active.scope.anchor
            if anchor is not None:
                root.add_branch_edge(anchor[0], anchor[1])

        else:
            self.active = None

        return dag, root, end

current_scope property

TraceScope currently active in the pipeline being traced.

Raises:

Type	Description
DAGNotSetError	No pipelines are being compiled.

Returns:

Name	Type	Description
TraceScope	TraceScope	Current scope.

__init__()

Initialize the compiler.

Source code in python/sdag/compiler.py

def __init__(self):
    """Initialize the compiler."""
    self.outer: list[CompiledDAG] = []
    self.active: CompiledDAG | None = None
    self._uid = 0

close_clause(branch_uid, next_edge_value, terminal=False)

Record a closed branch for the next Elif/Else.

Parameters:

Name	Type	Description	Default
branch_uid	int	Uid of the BranchNode the clause evaluated.	required
next_edge_value	bool	Branch edge value a following Elif/Else must depend on.	required
terminal	bool	True if this clause was an Else, closing the chain to further Elif/Else.	False

Source code in python/sdag/compiler.py

def close_clause(
    self,
    branch_uid: int,
    next_edge_value: bool,
    terminal: bool = False,  # noqa: FBT002
) -> None:
    """Record a closed branch for the next Elif/Else.

    Args:
        branch_uid (int): Uid of the BranchNode the clause evaluated.
        next_edge_value (bool): Branch edge value a following
            Elif/Else must depend on.
        terminal (bool): True if this clause was an Else, closing
            the chain to further Elif/Else.
    """
    self.current_scope.last_clause = ClosedClause(
        branch_uid=branch_uid,
        next_edge_value=next_edge_value,
        terminal=terminal,
    )

get_dag()

Detach a dag from the compiler.

Raises:

Type	Description
DAGNotSetError	No DAGs are set.

Returns:

Type	Description
tuple[DAG, RootNode, EndNode]	tuple[DAG, RootNode, EndNode]: Graph, root, and endnode.

Source code in python/sdag/compiler.py

def get_dag(self) -> tuple[DAG, RootNode, EndNode]:
    """Detach a dag from the compiler.

    Raises:
        DAGNotSetError: No DAGs are set.

    Returns:
        tuple[DAG, RootNode, EndNode]: Graph, root, and endnode.
    """
    if self.active is None:
        raise DAGNotSetError

    dag = self.active.dag
    root = self.active.root
    end = self.active.end
    parent_uids = set()

    for node in dag.nodes:
        if not node.parents and node.uid not in (root.uid, end.uid):
            node.add_logical_edge(root.uid)
        for parent in node.parents:
            parent_uids.add(parent.uid)

    for node in dag.nodes:
        if node.uid not in parent_uids:
            end.add_logical_edge(node.uid)
    self.register(end)

    if self.outer:
        self.active = self.outer.pop()
        self.active.dag.nodes.extend(dag.nodes)

        anchor = self.active.scope.anchor
        if anchor is not None:
            root.add_branch_edge(anchor[0], anchor[1])

    else:
        self.active = None

    return dag, root, end

get_uid()

Get a new uid.

Returns:

Name	Type	Description
int	int	Unique id.

Source code in python/sdag/compiler.py

def get_uid(self) -> int:
    """Get a new uid.

    Returns:
        int: Unique id.
    """
    uid = self._uid
    self._uid += 1
    return uid

pop_scope()

Exit the current lexical scope, restoring its parent.

Raises:

Type	Description
DAGNotSetError	No pipelines are being compiled, or attempting to pop the pipeline's root scope.

Returns:

Name	Type	Description
TraceScope	TraceScope	The scope that was just closed.

Source code in python/sdag/compiler.py

def pop_scope(self) -> TraceScope:
    """Exit the current lexical scope, restoring its parent.

    Raises:
        DAGNotSetError: No pipelines are being compiled, or
            attempting to pop the pipeline's root scope.

    Returns:
        TraceScope: The scope that was just closed.
    """
    closed = self.current_scope
    if closed.parent is None:
        raise DAGNotSetError
    self.active.scope = closed.parent  # type: ignore
    return closed

push_scope(anchor)

Enter a new lexical scope, e.g. an If/Elif/Else body.

Parameters:

Name	Type	Description	Default
anchor	tuple[int, bool] | None	(branch_uid, branch_value) stamped on every node registered in the new scope.	required

Returns:

Name	Type	Description
TraceScope	TraceScope	The newly active scope.

Source code in python/sdag/compiler.py

def push_scope(self, anchor: tuple[int, bool] | None) -> TraceScope:
    """Enter a new lexical scope, e.g. an If/Elif/Else body.

    Args:
        anchor (tuple[int, bool] | None): (branch_uid, branch_value)
            stamped on every node registered in the new scope.

    Returns:
        TraceScope: The newly active scope.
    """
    scope = TraceScope(parent=self.current_scope, anchor=anchor)
    self.active.scope = scope  # type: ignore
    return scope

register(node)

Register a node in the graph.

If a branch scope is currently active, the node is given a branch edge to it, tying it to that branch's true/false path.

Parameters:

Name	Type	Description	Default
node	NodeUnion	Node.	required

Raises:

Type	Description
DAGNotSetError	Attempting to register a node but no pipelines are being compiled.

Source code in python/sdag/compiler.py

def register(self, node: NodeUnion) -> None:
    """Register a node in the graph.

    If a branch scope is currently active, the node is given a
    branch edge to it, tying it to that branch's true/false path.

    Args:
        node (NodeUnion): Node.

    Raises:
        DAGNotSetError: Attempting to register a node
            but no pipelines are being compiled.
    """
    if self.active is None:
        raise DAGNotSetError

    if not node.pipeline_name:
        name = self.active.dag.meta.pipeline_name
        node.pipeline_name = name
        node.root_uid = self.active.root.uid

    self.active.dag.nodes.append(node)

    anchor = self.active.scope.anchor
    if anchor is not None:
        node.add_branch_edge(anchor[0], anchor[1])

reset()

Reset the compiler state.

Called before every compiled pipeline.

Source code in python/sdag/compiler.py

def reset(self) -> None:
    """Reset the compiler state.

    Called before every compiled pipeline.
    """
    self.outer = []
    self.active = None
    self._uid = 0

set_dag(root, import_path=None)

Set a new dag.

Every time a new pipeline is called.

Parameters:

Name	Type	Description	Default
root	RootNode	Pipeline root node.	required
import_path	str | None	Pipeline import path. If null, it defaults to the pipeline name. Defaults to None.	None

Source code in python/sdag/compiler.py

def set_dag(self, root: RootNode, import_path: str | None = None) -> None:
    """Set a new dag.

    Every time a new pipeline is called.

    Args:
        root (RootNode): Pipeline root node.
        import_path (str | None, optional): Pipeline import path.
            If null, it defaults to the pipeline name.
            Defaults to None.
    """
    name = root.pipeline_name
    import_path = import_path if import_path is not None else name
    dag = DAG(meta=DAGMeta(pipeline_name=name, import_path=import_path))
    end = EndNode(
        uid=self.get_uid(), root_uid=root.uid, pipeline_name=name
    )
    active = CompiledDAG(dag=dag, root=root, end=end)

    if self.active is not None:
        self.outer.append(self.active)

    self.active = active
    self.register(root)

SDAG

Master.

Keeps track of the imported pipelines and tasks. An error is raised is two tasks with the same name are imported.

Attributes:

Name	Type	Description
tasks	dict[str, Task]	Imported (global) tasks.
pipelines	dict[str, Pipeline]	Imported pipelines.

Source code in python/sdag/compiler.py

class SDAG:
    """Master.

    Keeps track of the imported pipelines and tasks.
    An error is raised is two tasks with the same name
    are imported.

    Attributes:
        tasks (dict[str, Task]): Imported (global) tasks.
        pipelines (dict[str, Pipeline]): Imported pipelines.
    """

    def __init__(self):
        """Initialize the master."""
        self.tasks: dict[str, Task] = {}
        self.pipelines: dict[str, Pipeline] = {}

    def add_pipeline(self, pipeline: "Pipeline") -> None:
        """Add a pipeline.

        Args:
            pipeline (Pipeline): Pipeline imported.
        """
        self.pipelines[pipeline.fn.__name__] = pipeline

    def add_task(self, task: "Task") -> None:
        """Add a task.

        Args:
            task (Task): Add a task.

        Raises:
            TaskNotUniqueError: The task is not unique.
        """
        if task.fn.__name__ in self.tasks:
            raise TaskNotUniqueError(name=task.fn.__name__)
        self.tasks[task.fn.__name__] = task

    def _reset(self) -> None:
        self.tasks = {}
        self.pipelines = {}

__init__()

Initialize the master.

Source code in python/sdag/compiler.py

def __init__(self):
    """Initialize the master."""
    self.tasks: dict[str, Task] = {}
    self.pipelines: dict[str, Pipeline] = {}

add_pipeline(pipeline)

Add a pipeline.

Parameters:

Name	Type	Description	Default
pipeline	Pipeline	Pipeline imported.	required

Source code in python/sdag/compiler.py

def add_pipeline(self, pipeline: "Pipeline") -> None:
    """Add a pipeline.

    Args:
        pipeline (Pipeline): Pipeline imported.
    """
    self.pipelines[pipeline.fn.__name__] = pipeline

add_task(task)

Add a task.

Parameters:

Name	Type	Description	Default
task	Task	Add a task.	required

Raises:

Type	Description
TaskNotUniqueError	The task is not unique.

Source code in python/sdag/compiler.py

def add_task(self, task: "Task") -> None:
    """Add a task.

    Args:
        task (Task): Add a task.

    Raises:
        TaskNotUniqueError: The task is not unique.
    """
    if task.fn.__name__ in self.tasks:
        raise TaskNotUniqueError(name=task.fn.__name__)
    self.tasks[task.fn.__name__] = task

constants

POSIX_ENV_VARIABLES = ['ARFLAGS', 'IFS', 'MAILPATH', 'PS1', 'CC', 'LANG', 'MAILRC', 'PS2', 'CDPATH', 'LC_ALL', 'MAKEFLAGS', 'PS3', 'CFLAGS', 'LC_COLLATE', 'MAKESHELL', 'PS4', 'CHARSET', 'LC_CTYPE', 'MANPATH', 'PWD', 'COLUMNS', 'LC_MESSAGES', 'MBOX', 'RANDOM', 'DATEMSK', 'LC_MONETARY', 'MORE', 'SECONDS', 'DEAD', 'LC_NUMERIC', 'MSGVERB', 'SHELL', 'EDITOR', 'LC_TIME', 'NLSPATH', 'TERM', 'ENV', 'LDFLAGS', 'NPROC', 'TERMCAP', 'EXINIT', 'LEX', 'OLDPWD', 'TERMINFO', 'FC', 'LFLAGS', 'OPTARG', 'TMPDIR', 'FCEDIT', 'LINENO', 'OPTERR', 'TZ', 'FFLAGS', 'LINES', 'OPTIND', 'USER', 'GET', 'LISTER', 'PAGER', 'VISUAL', 'GFLAGS', 'LOGNAME', 'PATH', 'YACC', 'HISTFILE', 'LPDEST', 'PPID', 'YFLAGS', 'HISTORY', 'MAIL', 'PRINTER', 'HISTSIZE', 'MAILCHECK', 'PROCLANG', 'HOME', 'MAILER', 'PROJECTDIR'] module-attribute

POSIX environment variables, taken from: https://pubs.opengroup.org/onlinepubs/9699919799/basedefs/V1_chap08.html.

It would be dangerous to override these variables with user-defined input arguments. So, if any of these variables is used as input for an external task, an exception is thrown.

core

__all__ = ['run', 'restart_run', 'kill_run', 'prune_cache', 'view_pipeline', 'run_single_task', 'print_pipeline_list', 'print_runs', 'describe_pipeline', 'print_status'] module-attribute

Built-in mutable sequence.

If no argument is given, the constructor creates a new empty list. The argument must be an iterable if specified.

__doc__ = '' module-attribute

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.

__file__ = '/leonardo_scratch/fast/BOOST_EUROPA/arognoni/tests/sdag/python/sdag/core.abi3.so' module-attribute

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.

__name__ = 'python.sdag.core' module-attribute

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.

__package__ = 'python.sdag' module-attribute

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.

detach

Job detaching.

detach_command()

Detach an sdag command.

Source code in python/sdag/detach.py

def detach_command() -> None:
    """Detach an sdag command."""
    pyproj = get_pyproj()
    configure_logging(log_level=pyproj.log_level)

    logger.info("Detaching command 'sdag %s'", " ".join(sys.argv[1:]))
    args = _filter_out_detach()
    _run_detached_command(args, pyproj)

discovery

Pipeline discovery and lazy import.

Pipelines are discovered via AST parsing, so we don't have to explicitely import files.

add_cwd_to_path()

Add the cwd to the path.

Useful for path.to.module:dag imports.

Source code in python/sdag/discovery.py

def add_cwd_to_path() -> None:
    """Add the cwd to the path.

    Useful for path.to.module:dag imports.
    """
    cwd = Path.cwd()
    sys.path.insert(0, str(cwd))

find_all_pipelines()

Find all importable pipelines.

Returns:

Type	Description
dict[str, str]	dict[str, str]: Pipeline names and paths.

Source code in python/sdag/discovery.py

def find_all_pipelines() -> dict[str, str]:
    """Find all importable pipelines.

    Returns:
        dict[str, str]: Pipeline names and paths.
    """
    pipelines: dict[str, str] = {}
    dag_dir = _find_dag_dir()
    for path in Path(dag_dir).rglob("**/*.py"):
        tree = ast.parse(path.read_text())
        for node in ast.walk(tree):
            if isinstance(node, ast.FunctionDef):
                for d in node.decorator_list:
                    name = _extract_decorator_name(d)
                    if name == "pipeline":
                        pipelines[node.name] = str(path)
    return pipelines

find_pipeline_by_name(name)

Import a pipeline from its name.

Parameters:

Name	Type	Description	Default
name	str	If the name is provided, the pipeline is discovered and imported. If the input path in the form path.to.module:pipeline is used, discovery is skipped and the pipeline is imported directly.	required

Raises:

Type	Description
DAGNotFoundError	The DAG is not found in the path.

Returns:

Name	Type	Description
Pipeline	Pipeline	Imported pipeline.

Source code in python/sdag/discovery.py

def find_pipeline_by_name(name: str) -> Pipeline:
    """Import a pipeline from its name.

    Args:
        name (str): If the name is provided, the
            pipeline is discovered and imported. If the
            input path in the form `path.to.module:pipeline`
            is used, discovery is skipped and the pipeline is
            imported directly.

    Raises:
        DAGNotFoundError: The DAG is not found in the path.

    Returns:
        Pipeline: Imported pipeline.
    """
    splits = name.split(":")
    pipeline_name = splits[-1]
    if pipeline_name in master.pipelines:
        return master.pipelines[pipeline_name]

    if ":" in name:
        module_path = "".join(splits[:-1])
    else:
        module_path = _find_module_path(pipeline_name)

    logger.info(
        "Importing pipeline '%s' from module '%s'", pipeline_name, module_path
    )
    try:
        module = importlib.import_module(module_path)
        pipeline = getattr(module, pipeline_name)
    except (ImportError, AttributeError) as exc:
        msg = f"Pipeline '{name}' not found"
        raise DAGNotFoundError(msg) from exc

    return pipeline

get_importable_module_path(path)

Turn a module path into its import string.

Parameters:

Name	Type	Description	Default
path	Path	Module path.	required

Returns:

Name	Type	Description
str	str	Import string.

Source code in python/sdag/discovery.py

def get_importable_module_path(path: Path) -> str:
    """Turn a module path into its import string.

    Args:
        path (Path): Module path.

    Returns:
        str: Import string.
    """
    file_path = path.resolve()
    root = _detect_root(file_path)
    root = root.resolve()

    logger.debug("Detected root: '%s'", root)
    relative = file_path.relative_to(root)
    parts = relative.with_suffix("").parts

    return ".".join(parts)

entrypoints

sdag entrypoints.

cli()

CLI entrypoint.

Source code in python/sdag/entrypoints.py

def cli():
    """CLI entrypoint."""
    from sdag.detach import detach_command
    from sdag.exceptions import CLIError
    from sdag.parser import ExtraArgumentParser, ParserBuilder

    add_cwd_to_path()
    parser = (
        ParserBuilder()
        .add_version()
        .add_compile_subparser()
        .add_run_subparser()
        .add_restart_subparser()
        .add_retry_subparser()
        .add_kill_subparser()
        .add_prune_subparser()
        .add_runtask_subparser()
        .add_view_subparser()
        .add_list_subparser()
        .add_describe_subparser()
        .add_status_subparser()
        .get_parser()
    )

    args, extras = parser.parse_known_args()
    extra_parser = ExtraArgumentParser()
    extra_args = extra_parser.parse(extras)
    if "log_level" not in args or "fn" not in args:
        raise CLIError

    if args.__dict__.get("detach", False):
        detach_command()
        return

    args.fn(args, extra_args)

sdag_execute(configure_logger=True)

Task entry point.

This is the function called by script to run wrapping tasks. It is also possible to use the sdag-execute command.

Parameters:

Name	Type	Description	Default
configure_logger	bool	Whether to configure the default logger. Defaults to True.	True

Source code in python/sdag/entrypoints.py

def sdag_execute(configure_logger: bool = True) -> None:  # noqa: FBT002
    """Task entry point.

    This is the function called by script to run wrapping tasks.
    It is also possible to use the `sdag-execute` command.

    Args:
        configure_logger (bool): Whether to configure the default
            logger. Defaults to True.
    """
    import asyncio
    import inspect
    import json
    from time import perf_counter

    from sdag.io import IOHandler, find_and_import_task
    from sdag.settings import get_cached_settings

    add_cwd_to_path()
    start = perf_counter()
    settings = get_cached_settings()

    if configure_logger:
        configure_logging(settings.sdag_log_level)

    logger.info(
        "Task: '%s'\nFunction: '%s'\nNode: '%s'"
        "\nPipeline: '%s'\nTry number: '%s'",
        settings.sdag_task_name,
        settings.sdag_task_fn,
        settings.sdag_uid,
        settings.sdag_subpipeline_name,
        settings.sdag_try_num,
    )

    if settings.is_job_array():
        logger.info(
            "Starting job '%s' of array across field '%s'."
            " Shards are collected in '%s'.",
            settings.slurm_array_task_id,
            settings.sdag_each_key,
            settings.sdag_shard_path,
        )

    handler = IOHandler(
        uid=settings.sdag_uid,
        pipeline_dir=settings.sdag_pipeline_dir,
        shard_path=settings.sdag_shard_path,
        task_id=settings.slurm_array_task_id,
        each_key=settings.sdag_each_key,
    )

    task = find_and_import_task(
        task_name=settings.sdag_task_fn,
        pipeline_name=settings.sdag_pipeline_name,
        subpipeline_name=settings.sdag_subpipeline_name,
        import_path=settings.sdag_import_path,
        input_kwargs=settings.sdag_input_kwargs,
    )
    logger.info("Found task '%s'", task.name)
    sig = inspect.signature(task.fn)
    if settings.is_job_array():
        input_kwargs = handler.get_array_input(sig)
    else:
        input_kwargs = handler.get_input(sig)

    logger.info("Input values:\n%s", json.dumps(input_kwargs, indent=4))
    artifacts = handler.get_artifacts(sig, input_kwargs)
    logger.info("Artifacts:\n%s", handler.serialize_artifacts(artifacts))

    handler.cast_values(sig, input_kwargs)
    logger.info("Starting task execution")

    if inspect.iscoroutinefunction(task.fn):
        output = asyncio.run(task.fn(**input_kwargs))
    else:
        output = task.fn(**input_kwargs)

    logger.info("Output values:\n%s", json.dumps(output, indent=4))

    if settings.is_job_array():
        handler.serialize_array_output(output, artifacts)
    else:
        handler.serialize_output(output, artifacts)

    end = perf_counter()
    logger.info(
        "Task '%s' completed. Elapsed time: %s s",
        settings.sdag_task_name,
        round(end - start, 2),
    )

exceptions

Custom exceptions.

CLIError

Bases: SDAGError

CLI error.

Source code in python/sdag/exceptions.py

class CLIError(SDAGError):
    """CLI error."""

    def __init__(self):
        """Raise the exception."""
        msg = "Unknown CLI command."
        super().__init__(msg)

__init__()

Raise the exception.

Source code in python/sdag/exceptions.py

def __init__(self):
    """Raise the exception."""
    msg = "Unknown CLI command."
    super().__init__(msg)

DAGNotFoundError

Bases: SDAGError

DAG not found in path.

Source code in python/sdag/exceptions.py

class DAGNotFoundError(SDAGError):
    """DAG not found in path."""

DAGNotSetError

Bases: SDAGError

No DAG is set in the compiler.

Source code in python/sdag/exceptions.py

class DAGNotSetError(SDAGError):
    """No DAG is set in the compiler."""

    def __init__(self):
        """Raise the exception."""
        msg = "no pipelines are currently being compiled."
        super().__init__(msg)

__init__()

Raise the exception.

Source code in python/sdag/exceptions.py

def __init__(self):
    """Raise the exception."""
    msg = "no pipelines are currently being compiled."
    super().__init__(msg)

DynamicArtifactError

Bases: SDAGError

Dynamic artifacts are not allowed.

Source code in python/sdag/exceptions.py

class DynamicArtifactError(SDAGError):
    """Dynamic artifacts are not allowed."""

    def __init__(self, key: str):
        """Raise the error.

        Args:
            key (str): Input key not found.
        """
        msg = f"key '{key}': Dynamic artifacts are not allowed."
        super().__init__(msg)

__init__(key)

Raise the error.

Parameters:

Name	Type	Description	Default
key	str	Input key not found.	required

Source code in python/sdag/exceptions.py

def __init__(self, key: str):
    """Raise the error.

    Args:
        key (str): Input key not found.
    """
    msg = f"key '{key}': Dynamic artifacts are not allowed."
    super().__init__(msg)

ExtraCLIArgsError

Bases: SDAGError

Extra CLI arguments are not allowed.

Source code in python/sdag/exceptions.py

class ExtraCLIArgsError(SDAGError):
    """Extra CLI arguments are not allowed."""

    def __init__(self, extra_kwargs: dict[str, Any]):
        """Raise the exception.

        Args:
            extra_kwargs (dict[str, Any]): Extra args used.
        """
        msg = (
            "Extra CLI arguments are not allowed."
            f" Extra arguments used: {extra_kwargs}"
        )
        super().__init__(msg)

__init__(extra_kwargs)

Raise the exception.

Parameters:

Name	Type	Description	Default
extra_kwargs	dict[str, Any]	Extra args used.	required

Source code in python/sdag/exceptions.py

def __init__(self, extra_kwargs: dict[str, Any]):
    """Raise the exception.

    Args:
        extra_kwargs (dict[str, Any]): Extra args used.
    """
    msg = (
        "Extra CLI arguments are not allowed."
        f" Extra arguments used: {extra_kwargs}"
    )
    super().__init__(msg)

InconsistentArraySettingsError

Bases: SDAGError

Job array settings are not consistent.

Shard path and task id must be either both None or set to a valid value.

Source code in python/sdag/exceptions.py

class InconsistentArraySettingsError(SDAGError):
    """Job array settings are not consistent.

    Shard path and task id must be either both None or set to a
    valid value.
    """

    def __init__(
        self, shard_path: Path | None, task_id: int | None, each: str | None
    ) -> None:
        """Raise the exception.

        Args:
            shard_path (Path | None): Shard input path.
            task_id (int | None): Slurm array task id.
            each (str | None): Sharded key.
        """
        msg = (
            f"Shard path '{shard_path}', task array {task_id},"
            f" and sharded field '{each}' are not consistent"
        )
        super().__init__(msg)

__init__(shard_path, task_id, each)

Raise the exception.

Parameters:

Name	Type	Description	Default
shard_path	Path | None	Shard input path.	required
task_id	int | None	Slurm array task id.	required
each	str | None	Sharded key.	required

Source code in python/sdag/exceptions.py

def __init__(
    self, shard_path: Path | None, task_id: int | None, each: str | None
) -> None:
    """Raise the exception.

    Args:
        shard_path (Path | None): Shard input path.
        task_id (int | None): Slurm array task id.
        each (str | None): Sharded key.
    """
    msg = (
        f"Shard path '{shard_path}', task array {task_id},"
        f" and sharded field '{each}' are not consistent"
    )
    super().__init__(msg)

IncorrectElifError

Bases: SDAGError

Incorrect Elif expression.

Source code in python/sdag/exceptions.py

class IncorrectElifError(SDAGError):
    """Incorrect Elif expression."""

    def __init__(self):
        """Raise the exception."""
        msg = "Incorrect Elif expression."
        super().__init__(msg)

__init__()

Raise the exception.

Source code in python/sdag/exceptions.py

def __init__(self):
    """Raise the exception."""
    msg = "Incorrect Elif expression."
    super().__init__(msg)

IncorrectElseError

Bases: SDAGError

Incorrect else expression.

Source code in python/sdag/exceptions.py

class IncorrectElseError(SDAGError):
    """Incorrect else expression."""

    def __init__(self):
        """Raise the exception."""
        msg = "Incorrect Else expression."
        super().__init__(msg)

__init__()

Raise the exception.

Source code in python/sdag/exceptions.py

def __init__(self):
    """Raise the exception."""
    msg = "Incorrect Else expression."
    super().__init__(msg)

KwargNotFoundError

Bases: SDAGError

Task input kwarg not found.

Source code in python/sdag/exceptions.py

class KwargNotFoundError(SDAGError):
    """Task input kwarg not found."""

    def __init__(self, key: str):
        """Raise the error.

        Args:
            key (str): Input key not found.
        """
        msg = f"Input argument '{key}' not found."
        super().__init__(msg)

__init__(key)

Raise the error.

Parameters:

Name	Type	Description	Default
key	str	Input key not found.	required

Source code in python/sdag/exceptions.py

def __init__(self, key: str):
    """Raise the error.

    Args:
        key (str): Input key not found.
    """
    msg = f"Input argument '{key}' not found."
    super().__init__(msg)

MultipleEachError

Bases: SDAGError

Job arrays across multiple axes is not supported.

Source code in python/sdag/exceptions.py

class MultipleEachError(SDAGError):
    """Job arrays across multiple axes is not supported."""

    def __init__(self, key: str, old_key: str) -> None:
        """Raise the exception.

        Args:
            key (str): New key of the job array.
            old_key (str): Old key of the job array.
        """
        msg = (
            f"Trying to set a job array across key '{key}' but"
            f" '{old_key}' is already registered. Use a for loop"
            " over the first key to set up grids."
        )
        super().__init__(msg)

__init__(key, old_key)

Raise the exception.

Parameters:

Name	Type	Description	Default
key	str	New key of the job array.	required
old_key	str	Old key of the job array.	required

Source code in python/sdag/exceptions.py

def __init__(self, key: str, old_key: str) -> None:
    """Raise the exception.

    Args:
        key (str): New key of the job array.
        old_key (str): Old key of the job array.
    """
    msg = (
        f"Trying to set a job array across key '{key}' but"
        f" '{old_key}' is already registered. Use a for loop"
        " over the first key to set up grids."
    )
    super().__init__(msg)

POSIXOverrideError

Bases: SDAGError

External tasks have variables conflicting with POSX.

Source code in python/sdag/exceptions.py

class POSIXOverrideError(SDAGError):
    """External tasks have variables conflicting with POSX."""

    def __init__(self, task_name: str, key: str):
        """Raise the exception.

        Args:
            task_name (str): Task violating the check.
            key (str): Forbidden key used.
        """
        banned = ", ".join(POSIX_ENV_VARIABLES)
        key_up = key.upper()
        msg = (
            f"External task '{task_name}' is trying to set the '{key}'"
            f" keyword argument. Please choose a different name as '{key_up}'"
            " would override a standard POSIX environment variable."
            f" The banned keywords are:\n{banned}"
        )
        super().__init__(msg)

__init__(task_name, key)

Raise the exception.

Parameters:

Name	Type	Description	Default
task_name	str	Task violating the check.	required
key	str	Forbidden key used.	required

Source code in python/sdag/exceptions.py

def __init__(self, task_name: str, key: str):
    """Raise the exception.

    Args:
        task_name (str): Task violating the check.
        key (str): Forbidden key used.
    """
    banned = ", ".join(POSIX_ENV_VARIABLES)
    key_up = key.upper()
    msg = (
        f"External task '{task_name}' is trying to set the '{key}'"
        f" keyword argument. Please choose a different name as '{key_up}'"
        " would override a standard POSIX environment variable."
        f" The banned keywords are:\n{banned}"
    )
    super().__init__(msg)

SDAGError

Bases: Exception

SDAG exception base class.

Source code in python/sdag/exceptions.py

class SDAGError(Exception):
    """SDAG exception base class."""

TaskNotUniqueError

Bases: SDAGError

Task name is not unique.

Source code in python/sdag/exceptions.py

class TaskNotUniqueError(SDAGError):
    """Task name is not unique."""

    def __init__(self, name: str):
        """Raise the error.

        Args:
            name (str): Not unique task name.
        """
        msg = f"Task name '{name}' is not unique."
        super().__init__(msg)

__init__(name)

Raise the error.

Parameters:

Name	Type	Description	Default
name	str	Not unique task name.	required

Source code in python/sdag/exceptions.py

def __init__(self, name: str):
    """Raise the error.

    Args:
        name (str): Not unique task name.
    """
    msg = f"Task name '{name}' is not unique."
    super().__init__(msg)

io

Task I/O management.

IOHandler

Manage I/O.

Attributes:

Name	Type	Description
uid	int	Task uid.
_pipeline_fname		DAG filename.
_output_fname		Node output filename as specified in the scheduler.
_input_fname		Node output filename as specified in the scheduler.
_shard_fname		Sharded input filename as specified in the scheduler.

Source code in python/sdag/io.py

class IOHandler:
    """Manage I/O.

    Attributes:
        uid (int): Task uid.
        _pipeline_fname: DAG filename.
        _output_fname: Node output filename as specified in the
            scheduler.
        _input_fname: Node output filename as specified in the
            scheduler.
        _shard_fname: Sharded input filename as specified in the
            scheduler.
    """

    def __init__(
        self,
        uid: int,
        pipeline_dir: Path,
        shard_path: Path | None,
        task_id: int | None,
        each_key: str | None,
    ):
        """Initialize the I/O handler.

        Args:
            uid (int): Task unique UID.
            pipeline_dir (Path): Pipeline working directory.
            shard_path (Path | None): Shard path.
            task_id (int | None): Array task id.
            each_key (str | None): Sharded field in the array.
        """
        self.uid = uid
        self.pipeline_dir = pipeline_dir
        self.shard_path = shard_path
        self.task_id = task_id
        self.each_key = each_key
        self._pipeline_fname = "pipeline.json"
        self._output_fname = "output.json"
        self._input_fname = "input.json"
        self._shard_fname = "shard.json"

    def get_input(self, sig: inspect.Signature) -> dict[str, Any]:
        """Read the input data.

        It's written by the scheduler in the node directory.

        Args:
            sig (inspect.Signature): Task signature used to
                validate the input values.

        Returns:
            dict[str, Any]: Task input kwargs.
        """
        input_data = self._load_input()
        validate_kwargs(sig, input_data)
        return input_data

    def get_array_input(
        self,
        sig: inspect.Signature,
    ) -> dict[str, Any]:
        """Read the array shard input data.

        It's written by the scheduler in the node directory.

        Args:
            shard_path (Path): Shard base path.
            task_id (int): ID of the task in the array.
            each_key (str): Sharded key.
            sig (inspect.Signature): Task signature used to
                validate the input values.

        Returns:
            dict[str, Any]: Task input kwargs.
        """
        if (
            self.shard_path is None
            or self.each_key is None
            or self.task_id is None
        ):
            raise InconsistentArraySettingsError(
                shard_path=self.shard_path,
                task_id=self.task_id,
                each=self.each_key,
            )

        input_data = self._load_input()
        path = self.shard_path / f"{self.task_id}" / self._shard_fname
        with path.open() as f:
            shard_data: dict[str, Any] = json.load(f)
        input_data[self.each_key] = shard_data

        validate_kwargs(sig, input_data)
        return input_data

    def get_artifacts(
        self, sig: inspect.Signature, input_kwargs: dict[str, Any]
    ) -> dict[str, ArtifactEdge]:
        """Build artifacts for execution.

        Args:
            sig (inspect.Signature): Task function signature.
            input_kwargs (dict[str, Any]): Input data. Those
                corresponding to artifacts will be turned into
                paths.

        Returns:
            dict[str, ArtifactEdge]: Artifact keys and artifacts.
                They will be merged with the input_kwargs.
        """
        artifacts: dict[str, ArtifactEdge] = {}
        for key, param in sig.parameters.items():
            if is_artifact(param):
                path = input_kwargs.get(key, "")
                artifacts[key] = ArtifactEdge(name=key, path=Path(path))
        return artifacts

    def serialize_output(
        self, output: Any, artifacts: dict[str, ArtifactEdge]
    ) -> None:
        """Serialize the output as JSON.

        Args:
            output (Any): Node output.
            artifacts (dict[str, ArtifactEdge]): Output artifacts.
        """
        artifact_list = list(artifacts.values())
        task_output = TaskOutput(output=output, artifacts=artifact_list)
        data = task_output.model_dump_json()
        output_path = self.pipeline_dir / f"{self.uid}" / self._output_fname

        with output_path.open("w") as f:
            f.write(data)

    def serialize_array_output(
        self, output: Any, artifacts: dict[str, ArtifactEdge]
    ) -> None:
        """Serialize a shard of output as JSON.

        Args:
            output (Any): Node output.
            artifacts (dict[str, ArtifactEdge]): Output artifacts.
        """
        if (
            self.shard_path is None
            or self.each_key is None
            or self.task_id is None
        ):
            raise InconsistentArraySettingsError(
                shard_path=self.shard_path,
                task_id=self.task_id,
                each=self.each_key,
            )

        artifact_list = list(artifacts.values())
        task_output = TaskOutput(output=output, artifacts=artifact_list)
        data = task_output.model_dump_json()
        output_path = self.shard_path / f"{self.task_id}" / self._output_fname

        with output_path.open("w") as f:
            f.write(data)

    def cast_values(
        self, sig: inspect.Signature, input_kwargs: dict[str, Any]
    ) -> None:
        """Automatically cast to Path if required.

        Args:
            sig (inspect.Signature): Task signature.
            input_kwargs (dict[str, Any]): Input arguments.
        """
        for key, param in sig.parameters.items():
            if param.annotation is Path or (
                is_artifact(param) and param.annotation.__origin__ is Path
            ):
                input_kwargs[key] = Path(input_kwargs[key])

    def serialize_artifacts(self, artifacts: dict[str, ArtifactEdge]) -> str:
        """Serialize artifacts as string for logging.

        Args:
            artifacts (dict[str, ArtifactEdge]): Artifacts.

        Returns:
            str: Serialized artifacts.
        """
        if not artifacts:
            return "{}"

        return "\n".join(
            artifact.model_dump_json(indent=4)
            for artifact in artifacts.values()
        )

    def _load_input(self) -> dict[str, Any]:
        """Load input without validation.

        Returns:
            dict[str, Any]: Loaded input
        """
        input_path = self.pipeline_dir / f"{self.uid}" / self._input_fname
        with input_path.open() as f:
            return json.load(f)

__init__(uid, pipeline_dir, shard_path, task_id, each_key)

Initialize the I/O handler.

Parameters:

Name	Type	Description	Default
uid	int	Task unique UID.	required
pipeline_dir	Path	Pipeline working directory.	required
shard_path	Path | None	Shard path.	required
task_id	int | None	Array task id.	required
each_key	str | None	Sharded field in the array.	required

Source code in python/sdag/io.py

def __init__(
    self,
    uid: int,
    pipeline_dir: Path,
    shard_path: Path | None,
    task_id: int | None,
    each_key: str | None,
):
    """Initialize the I/O handler.

    Args:
        uid (int): Task unique UID.
        pipeline_dir (Path): Pipeline working directory.
        shard_path (Path | None): Shard path.
        task_id (int | None): Array task id.
        each_key (str | None): Sharded field in the array.
    """
    self.uid = uid
    self.pipeline_dir = pipeline_dir
    self.shard_path = shard_path
    self.task_id = task_id
    self.each_key = each_key
    self._pipeline_fname = "pipeline.json"
    self._output_fname = "output.json"
    self._input_fname = "input.json"
    self._shard_fname = "shard.json"

cast_values(sig, input_kwargs)

Automatically cast to Path if required.

Parameters:

Name	Type	Description	Default
sig	Signature	Task signature.	required
input_kwargs	dict[str, Any]	Input arguments.	required

Source code in python/sdag/io.py

def cast_values(
    self, sig: inspect.Signature, input_kwargs: dict[str, Any]
) -> None:
    """Automatically cast to Path if required.

    Args:
        sig (inspect.Signature): Task signature.
        input_kwargs (dict[str, Any]): Input arguments.
    """
    for key, param in sig.parameters.items():
        if param.annotation is Path or (
            is_artifact(param) and param.annotation.__origin__ is Path
        ):
            input_kwargs[key] = Path(input_kwargs[key])

get_array_input(sig)

Read the array shard input data.

It's written by the scheduler in the node directory.

Parameters:

Name	Type	Description	Default
shard_path	Path	Shard base path.	required
task_id	int	ID of the task in the array.	required
each_key	str	Sharded key.	required
sig	Signature	Task signature used to validate the input values.	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Task input kwargs.

Source code in python/sdag/io.py

def get_array_input(
    self,
    sig: inspect.Signature,
) -> dict[str, Any]:
    """Read the array shard input data.

    It's written by the scheduler in the node directory.

    Args:
        shard_path (Path): Shard base path.
        task_id (int): ID of the task in the array.
        each_key (str): Sharded key.
        sig (inspect.Signature): Task signature used to
            validate the input values.

    Returns:
        dict[str, Any]: Task input kwargs.
    """
    if (
        self.shard_path is None
        or self.each_key is None
        or self.task_id is None
    ):
        raise InconsistentArraySettingsError(
            shard_path=self.shard_path,
            task_id=self.task_id,
            each=self.each_key,
        )

    input_data = self._load_input()
    path = self.shard_path / f"{self.task_id}" / self._shard_fname
    with path.open() as f:
        shard_data: dict[str, Any] = json.load(f)
    input_data[self.each_key] = shard_data

    validate_kwargs(sig, input_data)
    return input_data

get_artifacts(sig, input_kwargs)

Build artifacts for execution.

Parameters:

Name	Type	Description	Default
sig	Signature	Task function signature.	required
input_kwargs	dict[str, Any]	Input data. Those corresponding to artifacts will be turned into paths.	required

Returns:

Type	Description
dict[str, ArtifactEdge]	dict[str, ArtifactEdge]: Artifact keys and artifacts. They will be merged with the input_kwargs.

Source code in python/sdag/io.py

def get_artifacts(
    self, sig: inspect.Signature, input_kwargs: dict[str, Any]
) -> dict[str, ArtifactEdge]:
    """Build artifacts for execution.

    Args:
        sig (inspect.Signature): Task function signature.
        input_kwargs (dict[str, Any]): Input data. Those
            corresponding to artifacts will be turned into
            paths.

    Returns:
        dict[str, ArtifactEdge]: Artifact keys and artifacts.
            They will be merged with the input_kwargs.
    """
    artifacts: dict[str, ArtifactEdge] = {}
    for key, param in sig.parameters.items():
        if is_artifact(param):
            path = input_kwargs.get(key, "")
            artifacts[key] = ArtifactEdge(name=key, path=Path(path))
    return artifacts

get_input(sig)

Read the input data.

It's written by the scheduler in the node directory.

Parameters:

Name	Type	Description	Default
sig	Signature	Task signature used to validate the input values.	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Task input kwargs.

Source code in python/sdag/io.py

def get_input(self, sig: inspect.Signature) -> dict[str, Any]:
    """Read the input data.

    It's written by the scheduler in the node directory.

    Args:
        sig (inspect.Signature): Task signature used to
            validate the input values.

    Returns:
        dict[str, Any]: Task input kwargs.
    """
    input_data = self._load_input()
    validate_kwargs(sig, input_data)
    return input_data

serialize_array_output(output, artifacts)

Serialize a shard of output as JSON.

Parameters:

Name	Type	Description	Default
output	Any	Node output.	required
artifacts	dict[str, ArtifactEdge]	Output artifacts.	required

Source code in python/sdag/io.py

def serialize_array_output(
    self, output: Any, artifacts: dict[str, ArtifactEdge]
) -> None:
    """Serialize a shard of output as JSON.

    Args:
        output (Any): Node output.
        artifacts (dict[str, ArtifactEdge]): Output artifacts.
    """
    if (
        self.shard_path is None
        or self.each_key is None
        or self.task_id is None
    ):
        raise InconsistentArraySettingsError(
            shard_path=self.shard_path,
            task_id=self.task_id,
            each=self.each_key,
        )

    artifact_list = list(artifacts.values())
    task_output = TaskOutput(output=output, artifacts=artifact_list)
    data = task_output.model_dump_json()
    output_path = self.shard_path / f"{self.task_id}" / self._output_fname

    with output_path.open("w") as f:
        f.write(data)

serialize_artifacts(artifacts)

Serialize artifacts as string for logging.

Parameters:

Name	Type	Description	Default
artifacts	dict[str, ArtifactEdge]	Artifacts.	required

Returns:

Name	Type	Description
str	str	Serialized artifacts.

Source code in python/sdag/io.py

def serialize_artifacts(self, artifacts: dict[str, ArtifactEdge]) -> str:
    """Serialize artifacts as string for logging.

    Args:
        artifacts (dict[str, ArtifactEdge]): Artifacts.

    Returns:
        str: Serialized artifacts.
    """
    if not artifacts:
        return "{}"

    return "\n".join(
        artifact.model_dump_json(indent=4)
        for artifact in artifacts.values()
    )

serialize_output(output, artifacts)

Serialize the output as JSON.

Parameters:

Name	Type	Description	Default
output	Any	Node output.	required
artifacts	dict[str, ArtifactEdge]	Output artifacts.	required

Source code in python/sdag/io.py

def serialize_output(
    self, output: Any, artifacts: dict[str, ArtifactEdge]
) -> None:
    """Serialize the output as JSON.

    Args:
        output (Any): Node output.
        artifacts (dict[str, ArtifactEdge]): Output artifacts.
    """
    artifact_list = list(artifacts.values())
    task_output = TaskOutput(output=output, artifacts=artifact_list)
    data = task_output.model_dump_json()
    output_path = self.pipeline_dir / f"{self.uid}" / self._output_fname

    with output_path.open("w") as f:
        f.write(data)

find_and_import_task(task_name, pipeline_name, subpipeline_name, import_path, input_kwargs)

Sufficiently clever way to import tasks.

1. Check if the pipeline/task is already imported
2. import the pipeline with the import path
3. Try to import the subpipeline from its name
4. Compile the pipeline to import the subtask

Parameters:

Name	Type	Description	Default
task_name	str	Task name.	required
pipeline_name	str	Pipeline name.	required
subpipeline_name	str	Subpipeline name.	required
import_path	str	Pipeline import path.	required
input_kwargs	str	Input kwargs used to compile the pipeline serialized as a JSON string. Useful to recompile the pipeline.	required

Returns:

Name	Type	Description
Task	Task	Imported task.

Source code in python/sdag/io.py

def find_and_import_task(
    task_name: str,
    pipeline_name: str,
    subpipeline_name: str,
    import_path: str,
    input_kwargs: str,
) -> Task:
    """Sufficiently clever way to import tasks.

    1. Check if the pipeline/task is already imported
    2. import the pipeline with the import path
    3. Try to import the subpipeline from its name
    4. Compile the pipeline to import the subtask

    Args:
        task_name (str): Task name.
        pipeline_name (str): Pipeline name.
        subpipeline_name (str): Subpipeline name.
        import_path (str): Pipeline import path.
        input_kwargs (str): Input kwargs used to compile
            the pipeline serialized as a JSON string.
            Useful to recompile the pipeline.

    Returns:
        Task: Imported task.
    """
    logger.debug("Checking if the task is already imported")
    target_task = _get_already_imported_task(task_name, subpipeline_name)
    if target_task is not None:
        return target_task

    if pipeline_name == subpipeline_name:
        logger.debug("Task is not in an inner pipeline")
        target_task = _try_get_task(task_name, import_path)
        if target_task is not None:
            return target_task

    logger.debug("Trying to directly import the subpipeline")
    target_task = _try_get_task(task_name, subpipeline_name)
    if target_task is not None:
        return target_task

    logger.info(
        "Pipeline '%s' not found (might have been dynamically"
        " imported). Trying to compile the pipeline to find it",
        subpipeline_name,
    )

    kwargs = json.loads(input_kwargs)
    compile_and_return_dag(
        path=import_path, extra_metadata=None, input_kwargs=kwargs
    )
    target_task = _get_already_imported_task(task_name, subpipeline_name)
    if target_task is None:
        msg = f"Task '{task_name}' of pipeline '{subpipeline_name}' not found"
        raise DAGNotFoundError(msg) from None

    return target_task

mod_hashing

ModuleHasher

Hash modules.

Attributes:

Name	Type	Description
_excluded	list[Path]	Paths that must be excluded. They include stuff like builtins, third-party libraries etc.
_cwd	Path	Current working directory.

Source code in python/sdag/mod_hashing.py

class ModuleHasher:
    """Hash modules.

    Attributes:
        _excluded (list[Path]): Paths that must be excluded. They
            include stuff like builtins, third-party libraries etc.
        _cwd (Path): Current working directory.
    """

    def __init__(self) -> None:
        """Initialize the module hasher."""
        self._excluded = self._get_excluded_paths()
        self._cwd = Path.cwd().resolve()

    def hash_modules(self) -> str:
        """Hash internal modules.

        Returns:
            str: Computed hash. If no modules are found,
                an empty string is returned.
        """
        logger.info("Start module hashing")
        internal_modules = self._find_internal_modules()
        return self._compute_hash(internal_modules)

    def _find_internal_modules(self) -> list[Path]:
        """Find all internal modules.

        Returns:
            list[Path]: Internal modules. They will be read
                and their content hashed.
        """
        internal_modules: list[Path] = []
        for m in list(sys.modules.values()):
            module_file = getattr(m, "__file__", None)
            if module_file:
                path = Path(module_file).resolve()
                if path.is_file() and self._is_module_internal(path):
                    internal_modules.append(path)

        return internal_modules

    def _compute_hash(self, internal_modules: list[Path]) -> str:
        """Hash the content of all internal modules.

        They are sorted to enforce reproducibility.

        Args:
            internal_modules (list[Path]): Modules to be hashed.

        Returns:
            str: Hash.
        """
        import hashlib

        if not internal_modules:
            logger.warning("No internal modules found")
            return ""

        h = hashlib.sha256()
        for p in sorted(internal_modules):
            path_str = str(p)
            content = p.read_bytes()
            h.update(path_str.encode("utf-8"))
            h.update(content)
        return h.hexdigest()

    def _is_module_internal(self, module_path: Path) -> bool:
        """Check if a module is internal and thus must be hashed.

        Only Python modules in the cwd are considered.

        Args:
            module_path (Path): Module path to be checked.

        Returns:
            bool: True if the module is internal.
        """
        if module_path.suffix != ".py":
            return False

        if not module_path.is_relative_to(self._cwd):
            return False

        for excluded_path in self._excluded:
            if module_path.is_relative_to(excluded_path):
                return False
        return True

    def _get_excluded_paths(self) -> list[Path]:
        """Get the excluded path list.

        These are paths to builtin, 3rd party tools, etc.

        Returns:
            list[Path]: Excluded paths.
        """
        sysconfig_paths = sysconfig.get_paths()
        site_pkgs = site.getsitepackages()
        usersite_pkgs = site.getusersitepackages()
        paths = [
            *site_pkgs,
            usersite_pkgs,
            sysconfig_paths["stdlib"],
            sysconfig_paths["platstdlib"],
        ]

        if sys.prefix != sys.base_prefix:
            paths.append(sys.prefix)

        return [Path(p).resolve() for p in paths]

__init__()

Initialize the module hasher.

Source code in python/sdag/mod_hashing.py

def __init__(self) -> None:
    """Initialize the module hasher."""
    self._excluded = self._get_excluded_paths()
    self._cwd = Path.cwd().resolve()

hash_modules()

Hash internal modules.

Returns:

Name	Type	Description
str	str	Computed hash. If no modules are found, an empty string is returned.

Source code in python/sdag/mod_hashing.py

def hash_modules(self) -> str:
    """Hash internal modules.

    Returns:
        str: Computed hash. If no modules are found,
            an empty string is returned.
    """
    logger.info("Start module hashing")
    internal_modules = self._find_internal_modules()
    return self._compute_hash(internal_modules)

models

sdag models.

Artifact = Annotated[T, ArtifactSig] module-attribute

Artifact.

Artifacts are used to signal that some assets are being written. It is possible to use either Artifact[str] or Artifact[Path], they are automatically cast.

NodeUnion = Annotated[TaskNode | RootNode | EndNode | BranchNode | OneOfNode, Field(discriminator='kind')] module-attribute

Node union.

ParentKindUnion = Annotated[LogicalParent | OutputParent | ArtifactParent | BranchParent, Field(discriminator='kind')] module-attribute

Parent types.

ScriptUnion = Annotated[ScriptContent | ScriptPath, Field(discriminator='kind')] module-attribute

Script content or path.

ArtifactContainer

Artifact wrapper.

Hack to pass the node to the child alongside artifacts.

Attributes:

Name	Type	Description
key	str	Artifact key.
path	str	Artifact path.
node	Node	Parent node.

Source code in python/sdag/models.py

class ArtifactContainer:
    """Artifact wrapper.

    Hack to pass the node to the child alongside
    artifacts.

    Attributes:
        key (str): Artifact key.
        path (str): Artifact path.
        node (Node): Parent node.
    """

    def __init__(self, key: str, path: Path, node: "TaskNode"):
        """Initialize the artifact container.

        Args:
            key (str): Artifact key.
            path (str): Artifact path.
            node (Node): Parent node.
        """
        self.key = key
        self.path = path
        self.node = node

__init__(key, path, node)

Initialize the artifact container.

Parameters:

Name	Type	Description	Default
key	str	Artifact key.	required
path	str	Artifact path.	required
node	Node	Parent node.	required

Source code in python/sdag/models.py

def __init__(self, key: str, path: Path, node: "TaskNode"):
    """Initialize the artifact container.

    Args:
        key (str): Artifact key.
        path (str): Artifact path.
        node (Node): Parent node.
    """
    self.key = key
    self.path = path
    self.node = node

ArtifactEdge

Bases: BaseModel

Artifact edge.

Attributes:

Name	Type	Description
name	str	Artifact name.
path	Path	Artifact path.

Source code in python/sdag/models.py

class ArtifactEdge(BaseModel):
    """Artifact edge.

    Attributes:
        name (str): Artifact name.
        path (Path): Artifact path.
    """

    name: str
    path: Path

ArtifactParent

Bases: BaseModel

Artifact edge.

Attributes:

Name	Type	Description
kind	Literal['artifact']	Artifact kind.
key	str	Artifact key.
name	str	Artifact name.
path	str	Artifact path.

Source code in python/sdag/models.py

class ArtifactParent(BaseModel):
    """Artifact edge.

    Attributes:
        kind (Literal["artifact"]): Artifact kind.
        key (str): Artifact key.
        name (str): Artifact name.
        path (str): Artifact path.
    """

    kind: Literal["artifact"] = "artifact"
    key: str
    name: str
    path: Path

ArtifactSig

Used to mark artifacts.

Source code in python/sdag/models.py

class ArtifactSig:
    """Used to mark artifacts."""

BaseNode

Bases: BaseModel

Node base class.

Attributes:

Name	Type	Description
parents	list[Parent]	Parents.
uid	int	Unique id.
root_uid	int	Pipeline root unique id (added during compilation. Defaults to -1.
pipeline_name	str	Pipeline name, added during compilation. Defaults to ''.

Source code in python/sdag/models.py

class BaseNode(BaseModel):
    """Node base class.

    Attributes:
        parents (list[Parent]): Parents.
        uid (int): Unique id.
        root_uid (int): Pipeline root unique id (added
            during compilation. Defaults to -1.
        pipeline_name (str): Pipeline name, added during
            compilation. Defaults to ''.
    """

    parents: list[Parent] = Field(default_factory=list)
    uid: int
    root_uid: int = -1
    pipeline_name: str = ""

    def add_logical_edge(self, parent_uid: int) -> None:
        """Add a child->parent logical edge.

        It is a logical dependence, no data exchanged.

        Args:
            parent_uid (int): Parent uid.
        """
        parent = Parent(uid=parent_uid, kind=LogicalParent())
        self.parents.append(parent)

    def add_branch_edge(self, parent_uid: int, branch: bool) -> None:
        """Add a child -> parent branch edge.

        The child depends on one of the branches of an IfNode.

        Args:
            parent_uid (int): Parent uid.
            branch (bool): Branch the child depends on.
        """
        parent = Parent(uid=parent_uid, kind=BranchParent(branch=branch))
        self.parents.append(parent)

add_branch_edge(parent_uid, branch)

Add a child -> parent branch edge.

The child depends on one of the branches of an IfNode.

Parameters:

Name	Type	Description	Default
parent_uid	int	Parent uid.	required
branch	bool	Branch the child depends on.	required

Source code in python/sdag/models.py

def add_branch_edge(self, parent_uid: int, branch: bool) -> None:
    """Add a child -> parent branch edge.

    The child depends on one of the branches of an IfNode.

    Args:
        parent_uid (int): Parent uid.
        branch (bool): Branch the child depends on.
    """
    parent = Parent(uid=parent_uid, kind=BranchParent(branch=branch))
    self.parents.append(parent)

add_logical_edge(parent_uid)

Add a child->parent logical edge.

It is a logical dependence, no data exchanged.

Parameters:

Name	Type	Description	Default
parent_uid	int	Parent uid.	required

Source code in python/sdag/models.py

def add_logical_edge(self, parent_uid: int) -> None:
    """Add a child->parent logical edge.

    It is a logical dependence, no data exchanged.

    Args:
        parent_uid (int): Parent uid.
    """
    parent = Parent(uid=parent_uid, kind=LogicalParent())
    self.parents.append(parent)

BranchNode

Bases: BaseNode

Branch node.

Attributes:

Name	Type	Description
kind	Literal['branch']	Kind.

Source code in python/sdag/models.py

class BranchNode(BaseNode):
    """Branch node.

    Attributes:
        kind (Literal["branch"]): Kind.
    """

    kind: Literal["branch"] = "branch"

BranchParent

Bases: BaseModel

Branch edge.

Attributes:

Name	Type	Description
kind	Literal['branch']	Kind.
branch	bool	Branch.

Source code in python/sdag/models.py

class BranchParent(BaseModel):
    """Branch edge.

    Attributes:
        kind (Literal["branch"]): Kind.
        branch (bool): Branch.
    """

    kind: Literal["branch"] = "branch"
    branch: bool

CacheableTask

Bases: BaseModel

Cacheable task.

Info about tasks with caching enabled.

Attributes:

Name	Type	Description
name	str	Task name.
pipeline	str	Pipeline name.
scope	Scope	Task scope.

Source code in python/sdag/models.py

class CacheableTask(BaseModel):
    """Cacheable task.

    Info about tasks with caching enabled.

    Attributes:
        name (str): Task name.
        pipeline (str): Pipeline name.
        scope (Scope): Task scope.
    """

    name: str
    pipeline: str
    scope: Scope

ClosedClause dataclass

A closed If/Elif/Else clause, kept so the next Elif/Else can chain.

Attributes:

Name	Type	Description
branch_uid	int	Uid of the BranchNode this clause evaluated.
next_edge_value	bool	Branch edge value a following Elif/Else must depend on to run.
terminal	bool	True once an Else has closed the chain, so no further Elif/Else may follow.

Source code in python/sdag/models.py

@dataclass
class ClosedClause:
    """A closed If/Elif/Else clause, kept so the next Elif/Else can chain.

    Attributes:
        branch_uid (int): Uid of the BranchNode this clause evaluated.
        next_edge_value (bool): Branch edge value a following Elif/Else
            must depend on to run.
        terminal (bool): True once an Else has closed the chain, so no
            further Elif/Else may follow.
    """

    branch_uid: int
    next_edge_value: bool
    terminal: bool = False

CompiledDAG

Bases: BaseModel

Pipeline being compiled.

Attributes:

Name	Type	Description
dag	DAG	DAG.
root	RootNode	Root node.
end	EndNode	End node.
scope	TraceScope	Lexical scope currently active while tracing this pipeline's body.

Source code in python/sdag/models.py

class CompiledDAG(BaseModel):
    """Pipeline being compiled.

    Attributes:
        dag (DAG): DAG.
        root (RootNode): Root node.
        end (EndNode): End node.
        scope (TraceScope): Lexical scope currently active while
            tracing this pipeline's body.
    """

    dag: DAG
    root: RootNode
    end: EndNode
    scope: TraceScope = Field(default_factory=TraceScope)

DAG

Bases: BaseModel

Compiled pipeline.

Attributes:

Name	Type	Description
meta	DAGMeta	DAG metadata.
nodes	list[NodeUnion]	Nodes.

Source code in python/sdag/models.py

class DAG(BaseModel):
    """Compiled pipeline.

    Attributes:
        meta (DAGMeta): DAG metadata.
        nodes (list[NodeUnion]): Nodes.
    """

    meta: DAGMeta
    nodes: list[NodeUnion] = Field(default_factory=list)

DAGMeta

Bases: BaseModel

DAG metadata.

Attributes:

Name	Type	Description
pipeline_name	str	Pipeline name.
timestamp	str	Timestamp.
import_path	str	Import path.
code_hash	str	Hash of all modules imported. It is added after pipeline compilation.
kwargs	dict[str, Any]	Pipeline input kwargs.
extra	str	Extra metadata.

Source code in python/sdag/models.py

class DAGMeta(BaseModel):
    """DAG metadata.

    Attributes:
        pipeline_name (str): Pipeline name.
        timestamp (str): Timestamp.
        import_path (str): Import path.
        code_hash (str): Hash of all modules imported. It
            is added after pipeline compilation.
        kwargs (dict[str, Any]): Pipeline input kwargs.
        extra (str): Extra metadata.
    """

    pipeline_name: str
    timestamp: str = Field(default_factory=datetime.now().isoformat)
    import_path: str = ""
    code_hash: str = ""
    kwargs: dict[str, Any] = Field(default_factory=dict)
    extra: dict[str, Any] = Field(default_factory=dict)

    @model_validator(mode="after")
    def set_path_to_pipeline_name_if_missing(self) -> Self:
        """Set the import path to the pipeline name if null.

        Returns:
            Self: Metadata.
        """
        if not self.import_path:
            self.import_path = self.pipeline_name
        return self

set_path_to_pipeline_name_if_missing()

Set the import path to the pipeline name if null.

Returns:

Name	Type	Description
Self	Self	Metadata.

Source code in python/sdag/models.py

@model_validator(mode="after")
def set_path_to_pipeline_name_if_missing(self) -> Self:
    """Set the import path to the pipeline name if null.

    Returns:
        Self: Metadata.
    """
    if not self.import_path:
        self.import_path = self.pipeline_name
    return self

Each

Signal job arrays.

Attributes:

Name	Type	Description
collection	list[Any] | TaskNode	Inner keyword argument.

Source code in python/sdag/models.py

class Each:
    """Signal job arrays.

    Attributes:
        collection (list[Any] | TaskNode): Inner keyword
            argument.
    """

    def __init__(self, collection: list[Any] | TaskNode):
        """Initialize the array collection.

        Args:
            collection (list[Any] | TaskNode): Inner collection.
                It can be either a static or dynamic list.
        """
        self.collection = collection

__init__(collection)

Initialize the array collection.

Parameters:

Name	Type	Description	Default
collection	list[Any] | TaskNode	Inner collection. It can be either a static or dynamic list.	required

Source code in python/sdag/models.py

def __init__(self, collection: list[Any] | TaskNode):
    """Initialize the array collection.

    Args:
        collection (list[Any] | TaskNode): Inner collection.
            It can be either a static or dynamic list.
    """
    self.collection = collection

EndNode

Bases: BaseNode

End node.

Attributes:

Name	Type	Description
kind	Literal['end']	Kind.
_artifact_container	dict[str, dict[str, ArtifactContainer]]	Used to make artifact edges.

Source code in python/sdag/models.py

class EndNode(BaseNode):
    """End node.

    Attributes:
        kind (Literal["end"]): Kind.
        _artifact_container (dict[str, dict[str, ArtifactContainer]]):
            Used to make artifact edges.
    """

    kind: Literal["end"] = "end"
    _artifact_container: dict[str, dict[str, ArtifactContainer]] = PrivateAttr(
        default_factory=dict
    )

    @property
    def artifacts(self) -> dict[str, dict[str, ArtifactContainer]]:
        """Artifacts.

        Returns:
            dict[str, dict[str, ArtifactContainer]]: Artifacts.
        """
        return self._artifact_container

    def join_artifacts(self, node: TaskNode | OneOfNode) -> None:
        """Join all artifacts of the pipeline.

        Args:
            node (TaskNode | OneOfNode): Node.
        """
        self._artifact_container[node.name] = node.artifacts

artifacts property

Artifacts.

Returns:

Type	Description
dict[str, dict[str, ArtifactContainer]]	dict[str, dict[str, ArtifactContainer]]: Artifacts.

join_artifacts(node)

Join all artifacts of the pipeline.

Parameters:

Name	Type	Description	Default
node	TaskNode | OneOfNode	Node.	required

Source code in python/sdag/models.py

def join_artifacts(self, node: TaskNode | OneOfNode) -> None:
    """Join all artifacts of the pipeline.

    Args:
        node (TaskNode | OneOfNode): Node.
    """
    self._artifact_container[node.name] = node.artifacts

Kwarg

Bases: BaseModel

Task input kwarg.

Attributes:

Name	Type	Description
key	str	Key.
value	Any	Value.

Source code in python/sdag/models.py

class Kwarg(BaseModel):
    """Task input kwarg.

    Attributes:
        key (str): Key.
        value (Any): Value.
    """

    key: str
    value: Any

LogicalParent

Bases: BaseModel

Logical parent kind.

Attributes:

Name	Type	Description
kind	Literal['logical']	Kind.

Source code in python/sdag/models.py

class LogicalParent(BaseModel):
    """Logical parent kind.

    Attributes:
        kind (Literal["logical"]): Kind.
    """

    kind: Literal["logical"] = "logical"

OneOfNode

Bases: BaseNode

OneOf node.

Attributes:

Name	Type	Description
kind	Literal['one_of']	Kind.
name	str	Node name.
output_artifacts	list[ArtifactEdge]	Artifact edge.
_artifact_containers	dict[str, ArtifactContainer]	Used to make artifact edges.

Source code in python/sdag/models.py

class OneOfNode(BaseNode):
    """OneOf node.

    Attributes:
        kind (Literal["one_of"]): Kind.
        name (str): Node name.
        output_artifacts (list[ArtifactEdge]): Artifact edge.
        _artifact_containers (dict[str, ArtifactContainer]):
            Used to make artifact edges.
    """

    kind: Literal["one_of"] = "one_of"
    name: str
    output_artifacts: list[ArtifactEdge] = Field(
        default_factory=list, serialization_alias="artifacts"
    )
    _artifact_containers: dict[str, ArtifactContainer] = PrivateAttr(
        default_factory=dict
    )

    @property
    def artifacts(self) -> dict[str, ArtifactContainer]:
        """Get the artifacts.

        Returns:
            dict[str, ArtifactContainer]: Artifacts.
        """
        return self._artifact_containers

    def add_logical_edge(self, parent_uid: int) -> None:
        """Add a child->parent logical edge.

        It is a logical dependence, no data exchanged.

        Args:
            parent_uid (int): Parent uid.
        """
        parent = Parent(uid=parent_uid, kind=LogicalParent())
        self.parents.append(parent)

artifacts property

Get the artifacts.

Returns:

Type	Description
dict[str, ArtifactContainer]	dict[str, ArtifactContainer]: Artifacts.

add_logical_edge(parent_uid)

Add a child->parent logical edge.

It is a logical dependence, no data exchanged.

Parameters:

Name	Type	Description	Default
parent_uid	int	Parent uid.	required

Source code in python/sdag/models.py

def add_logical_edge(self, parent_uid: int) -> None:
    """Add a child->parent logical edge.

    It is a logical dependence, no data exchanged.

    Args:
        parent_uid (int): Parent uid.
    """
    parent = Parent(uid=parent_uid, kind=LogicalParent())
    self.parents.append(parent)

OutputParent

Bases: BaseModel

Output edge.

Attributes:

Name	Type	Description
kind	Literal['outout']	Kind.
key	str	Field name.

Source code in python/sdag/models.py

class OutputParent(BaseModel):
    """Output edge.

    Attributes:
        kind (Literal["outout"]): Kind.
        key (str): Field name.
    """

    kind: Literal["output"] = "output"
    key: str

Parent

Bases: BaseModel

Edge in the graph.

Attributes:

Name	Type	Description
uid	int	Parent type.
kind	str	Edge type.

Source code in python/sdag/models.py

class Parent(BaseModel):
    """Edge in the graph.

    Attributes:
        uid (int): Parent type.
        kind (str): Edge type.
    """

    uid: int
    kind: ParentKindUnion

RootNode

Bases: BaseNode

Root node.

Attributes:

Name	Type	Description
kind	Literal['root']	Kind.
root_uid	int	Same as uid.

Source code in python/sdag/models.py

class RootNode(BaseNode):
    """Root node.

    Attributes:
        kind (Literal["root"]): Kind.
        root_uid: Same as uid.
    """

    kind: Literal["root"] = "root"
    root_uid: int = Field(alias="uid", default=-1)

ScriptContent

Bases: BaseModel

Script embedded in the pipeline definition.

Attributes:

Name	Type	Description
kind	Literal['script']	Script kind.
content	str	Script content.

Source code in python/sdag/models.py

class ScriptContent(BaseModel):
    """Script embedded in the pipeline definition.

    Attributes:
        kind (Literal['script']): Script kind.
        content (str): Script content.
    """

    kind: Literal["script"] = "script"
    content: str

ScriptPath

Bases: BaseModel

Script path.

Attributes:

Name	Type	Description
kind	Literal['script_path']	Script path.
path	Path	Path.

Source code in python/sdag/models.py

class ScriptPath(BaseModel):
    """Script path.

    Attributes:
        kind (Literal['script_path']): Script path.
        path (Path): Path.
    """

    kind: Literal["script_path"] = "script_path"
    path: Path

SlurmOverride

Bases: BaseModel

Override Slurm arguments.

They must be set following the Slurm conventions.

Attributes:

Name	Type	Description
job_name	str | None	(str | None): Job name. Defaults to None.
nodes	NonNegativeInt | None	(int | None): Number of nodes. Defaults to None.
partition	str | None	(str | None): Partition. Defaults to None.
qos	str | None	(str | None) = qos. Defaults to None.
gpus_per_node	NonNegativeInt | None	(int | None): GPUs per node. Defaults to None.
ntasks_per_node	NonNegativeInt | None	(int | None): Tasks per node. Defaults to None.
output	str | None	(str | None): stdout. Defaults to None.
error	str | None	(str | None): stderr. Defaults to None.
account	str | None	(str | None): Account. Defaults to None.
cpus_per_task	int | None	CPUs per task. Defaults to None.
mem	str | None	(str | None): Memory. Defaults to None.
time	str | None	(str | None): Wall time. Defaults to None.

Source code in python/sdag/models.py

class SlurmOverride(BaseModel):
    """Override Slurm arguments.

    They must be set following the Slurm conventions.

    Attributes:
        job_name: (str | None): Job name. Defaults to None.
        nodes: (int | None): Number of nodes. Defaults to None.
        partition: (str | None): Partition. Defaults to None.
        qos: (str | None) = qos. Defaults to None.
        gpus_per_node: (int | None): GPUs per node. Defaults to None.
        ntasks_per_node: (int | None): Tasks per node.
            Defaults to None.
        output: (str | None): stdout. Defaults to None.
        error: (str | None): stderr. Defaults to None.
        account: (str | None): Account. Defaults to None.
        cpus_per_task (int | None): CPUs per task. Defaults to None.
        mem: (str | None): Memory. Defaults to None.
        time: (str | None): Wall time. Defaults to None.
    """

    job_name: str | None = Field(alias="job-name", default=None)
    nodes: NonNegativeInt | None = None
    partition: str | None = None
    qos: str | None = None
    gpus_per_node: NonNegativeInt | None = Field(
        alias="gpus-per-node", default=None
    )
    ntasks_per_node: NonNegativeInt | None = Field(
        alias="ntasks-per-node", default=None
    )
    output: str | None = None
    error: str | None = None
    account: str | None = None
    cpus_per_task: NonNegativeInt | None = Field(
        alias="cpus-per-task", default=None
    )
    mem: str | None = None
    time: str | None = None

    model_config = ConfigDict(
        extra="forbid", validate_by_alias=True, serialize_by_alias=True
    )

    def override_from(self, other: Self) -> None:  # noqa: C901
        """Override slurm configs from another set.

        Args:
            other (Self): Slurm configurations overriding
                the current ones in place.
        """
        if other.job_name is not None:
            self.job_name = other.job_name

        if other.nodes is not None:
            self.nodes = other.nodes

        if other.partition is not None:
            self.partition = other.partition

        if other.qos is not None:
            self.qos = other.qos

        if other.gpus_per_node is not None:
            self.gpus_per_node = other.gpus_per_node

        if other.ntasks_per_node is not None:
            self.ntasks_per_node = other.ntasks_per_node

        if other.output is not None:
            self.output = other.output

        if other.error is not None:
            self.error = other.error

        if other.account is not None:
            self.account = other.account

        if other.cpus_per_task is not None:
            self.cpus_per_task = other.cpus_per_task

        if other.mem is not None:
            self.mem = other.mem

        if other.time is not None:
            self.time = other.time

override_from(other)

Override slurm configs from another set.

Parameters:

Name	Type	Description	Default
other	Self	Slurm configurations overriding the current ones in place.	required

Source code in python/sdag/models.py

def override_from(self, other: Self) -> None:  # noqa: C901
    """Override slurm configs from another set.

    Args:
        other (Self): Slurm configurations overriding
            the current ones in place.
    """
    if other.job_name is not None:
        self.job_name = other.job_name

    if other.nodes is not None:
        self.nodes = other.nodes

    if other.partition is not None:
        self.partition = other.partition

    if other.qos is not None:
        self.qos = other.qos

    if other.gpus_per_node is not None:
        self.gpus_per_node = other.gpus_per_node

    if other.ntasks_per_node is not None:
        self.ntasks_per_node = other.ntasks_per_node

    if other.output is not None:
        self.output = other.output

    if other.error is not None:
        self.error = other.error

    if other.account is not None:
        self.account = other.account

    if other.cpus_per_task is not None:
        self.cpus_per_task = other.cpus_per_task

    if other.mem is not None:
        self.mem = other.mem

    if other.time is not None:
        self.time = other.time

TaskNode

Bases: BaseNode

Task.

Attributes:

Name	Type	Description
kind	Literal['task']	Kind.
parents	list[Parent]	Edges.
fn_name	str	Function name.
scope	Scope	Task scope.
name	str	Task name, linked to caching.
cache	CacheOptions	Caching options.
cache_ignore	list[str]	list of fields ignored during cache validation.
cache_size	int	Cache size. Ignore if caching is disabled. set to 0 to allow for infinite cache size. Defaults to 1.
code_hash	str	Task source code hash.
mode	ExecMode	Wrap a Python function or an external script.
cmd	Commands	Command used to launch the script.
retries	int	Retries.
each	str | None	If not null, it's the key of the array field. Defaults to None.
njobs	int	Number of jobs if it defines an array. Defaults to 1.
script	ScriptUnion	Script path or content.
tags	list[str]	Task tags.
kwargs	list[Kwarg]	Input kwargs.
envs	dict[str, Any]	Environment variables. They will be set in the task as strings.
slurm	SlurmOverride	Override Slurm resources.
output_artifacts	list[ArtifactEdge]	Artifacts.
_artifact_containers	dict[str, ArtifactContainer]	Used to add edges to the graph.

Source code in python/sdag/models.py

class TaskNode(BaseNode):
    """Task.

    Attributes:
        kind (Literal["task"]): Kind.
        parents (list[Parent]): Edges.
        fn_name (str): Function name.
        scope (Scope): Task scope.
        name (str): Task name, linked to caching.
        cache (CacheOptions): Caching options.
        cache_ignore (list[str]): list of fields ignored
            during cache validation.
        cache_size (int): Cache size. Ignore if caching is disabled.
            set to 0 to allow for infinite cache size. Defaults
            to 1.
        code_hash (str): Task source code hash.
        mode (ExecMode): Wrap a Python function or
            an external script.
        cmd (Commands): Command used to launch the script.
        retries (int): Retries.
        each (str | None): If not null, it's the key of
            the array field. Defaults to None.
        njobs (int): Number of jobs if it defines an
            array. Defaults to 1.
        script (ScriptUnion): Script path or content.
        tags (list[str]): Task tags.
        kwargs (list[Kwarg]): Input kwargs.
        envs (dict[str, Any]): Environment variables. They
            will be set in the task as strings.
        slurm (SlurmOverride): Override Slurm resources.
        output_artifacts (list[ArtifactEdge]): Artifacts.
        _artifact_containers (dict[str, ArtifactContainer]):
            Used to add edges to the graph.
    """

    kind: Literal["task"] = "task"
    parents: list[Parent] = Field(default_factory=list)
    fn_name: str
    name: str
    cache: CacheOptions
    scope: Scope
    cache_ignore: list[str] = Field(default_factory=list)
    cache_size: int = 1
    code_hash: str
    mode: ExecMode
    cmd: Commands
    retries: int
    each: str | None = None
    njobs: int = 1
    script: ScriptUnion
    tags: list[str] = Field(default_factory=list)
    kwargs: list[Kwarg] = Field(default_factory=list)
    envs: dict[str, Any] = Field(default_factory=dict)
    slurm: SlurmOverride = Field(default_factory=SlurmOverride)
    output_artifacts: list[ArtifactEdge] = Field(
        default_factory=list, serialization_alias="artifacts"
    )
    _artifact_containers: dict[str, ArtifactContainer] = PrivateAttr(
        default_factory=dict
    )

    @property
    def artifacts(self) -> dict[str, ArtifactContainer]:
        """Get the artifacts.

        Returns:
            dict[str, ArtifactContainer]: Artifacts.
        """
        return self._artifact_containers

    def register_artifact(self, key: str, path: Path) -> None:
        """Register an artifact.

        Args:
            key (str): Artifact key.
            path (Path): Artifact path.
        """
        self.output_artifacts.append(ArtifactEdge(name=key, path=path))
        self._artifact_containers[key] = ArtifactContainer(key, path, self)

    def add_output_edge(self, parent_uid, key: str) -> None:
        """Add a child->parent output edge.

        Child will read the parent output.

        Args:
            parent_uid (int): Parent uid.
            key (str): parent output key in the child input kwargs.
        """
        parent = Parent(uid=parent_uid, kind=OutputParent(key=key))
        self.parents.append(parent)

    def add_artifact_edge(
        self, parent_uid: int, key: str, name: str, path: Path
    ) -> None:
        """Add a child->parent artifact edge.

        The child will use an artifact produced by the parent.

        Args:
            parent_uid (int): Parent uid.
            key (str): Artifact key in the child input kwargs.
            name (str): Artifact name.
            path (Path): Artifact path.
        """
        parent = Parent(
            uid=parent_uid,
            kind=ArtifactParent(key=key, name=name, path=path),
        )
        self.parents.append(parent)

    def add_kwarg(self, kwarg: Kwarg) -> None:
        """Add a static kwarg.

        Args:
            kwarg (Kwarg): Kwarg to be added.

        Raises:
            POSIXOverrideError: Task is external and variable is a
                common POSIX variable.
        """
        if self.mode == "ext" and kwarg.key.upper() in POSIX_ENV_VARIABLES:
            raise POSIXOverrideError(task_name=self.name, key=kwarg.key)
        self.kwargs.append(kwarg)

artifacts property

Get the artifacts.

Returns:

Type	Description
dict[str, ArtifactContainer]	dict[str, ArtifactContainer]: Artifacts.

add_artifact_edge(parent_uid, key, name, path)

Add a child->parent artifact edge.

The child will use an artifact produced by the parent.

Parameters:

Name	Type	Description	Default
parent_uid	int	Parent uid.	required
key	str	Artifact key in the child input kwargs.	required
name	str	Artifact name.	required
path	Path	Artifact path.	required

Source code in python/sdag/models.py

def add_artifact_edge(
    self, parent_uid: int, key: str, name: str, path: Path
) -> None:
    """Add a child->parent artifact edge.

    The child will use an artifact produced by the parent.

    Args:
        parent_uid (int): Parent uid.
        key (str): Artifact key in the child input kwargs.
        name (str): Artifact name.
        path (Path): Artifact path.
    """
    parent = Parent(
        uid=parent_uid,
        kind=ArtifactParent(key=key, name=name, path=path),
    )
    self.parents.append(parent)

add_kwarg(kwarg)

Add a static kwarg.

Parameters:

Name	Type	Description	Default
kwarg	Kwarg	Kwarg to be added.	required

Raises:

Type	Description
POSIXOverrideError	Task is external and variable is a common POSIX variable.

Source code in python/sdag/models.py

def add_kwarg(self, kwarg: Kwarg) -> None:
    """Add a static kwarg.

    Args:
        kwarg (Kwarg): Kwarg to be added.

    Raises:
        POSIXOverrideError: Task is external and variable is a
            common POSIX variable.
    """
    if self.mode == "ext" and kwarg.key.upper() in POSIX_ENV_VARIABLES:
        raise POSIXOverrideError(task_name=self.name, key=kwarg.key)
    self.kwargs.append(kwarg)

add_output_edge(parent_uid, key)

Add a child->parent output edge.

Child will read the parent output.

Parameters:

Name	Type	Description	Default
parent_uid	int	Parent uid.	required
key	str	parent output key in the child input kwargs.	required

Source code in python/sdag/models.py

def add_output_edge(self, parent_uid, key: str) -> None:
    """Add a child->parent output edge.

    Child will read the parent output.

    Args:
        parent_uid (int): Parent uid.
        key (str): parent output key in the child input kwargs.
    """
    parent = Parent(uid=parent_uid, kind=OutputParent(key=key))
    self.parents.append(parent)

register_artifact(key, path)

Register an artifact.

Parameters:

Name	Type	Description	Default
key	str	Artifact key.	required
path	Path	Artifact path.	required

Source code in python/sdag/models.py

def register_artifact(self, key: str, path: Path) -> None:
    """Register an artifact.

    Args:
        key (str): Artifact key.
        path (Path): Artifact path.
    """
    self.output_artifacts.append(ArtifactEdge(name=key, path=path))
    self._artifact_containers[key] = ArtifactContainer(key, path, self)

TaskOutput

Bases: BaseModel

Output of a task.

Attributes:

Name	Type	Description
output	Any	Task output.
artifacts	list[ArtifactEdge]	Output artifacts.

Source code in python/sdag/models.py

class TaskOutput(BaseModel):
    """Output of a task.

    Attributes:
        output (Any): Task output.
        artifacts (list[ArtifactEdge]): Output artifacts.
    """

    output: Any
    artifacts: list[ArtifactEdge]

TraceScope dataclass

Branch scope.

Attributes:

Name	Type	Description
parent	Self | None	Enclosing scope. None for the pipeline's root scope. Defaults to None.
anchor	tuple[int, bool] | None	(branch_uid, branch_value) branch edge on every node registered while active. None outside of branches. Defaults to None.
last_clause	ClosedClause | None	Last branch closed directly in this scope, used by the next Elif/Else to chain onto it. Defaults to None.

Source code in python/sdag/models.py

@dataclass
class TraceScope:
    """Branch scope.

    Attributes:
        parent (Self | None): Enclosing scope. None for the
            pipeline's root scope. Defaults to None.
        anchor (tuple[int, bool] | None): (branch_uid, branch_value)
            branch edge on every node registered while active.
            None outside of branches. Defaults to None.
        last_clause (ClosedClause | None): Last branch closed
            directly in this scope, used by the next Elif/Else
            to chain onto it. Defaults to None.
    """

    parent: Self | None = None
    anchor: tuple[int, bool] | None = None
    last_clause: ClosedClause | None = None

parser

sdag parser.

ExtraArgumentParser

Parse the CLI extra arguments.

These arguments will become the pipline input kwargs. The parse_known_args method doesn't handle --key=value and --key value automatically, so some care must be taken.

Keep in mind that values are casted with json.

Source code in python/sdag/parser.py

class ExtraArgumentParser:
    """Parse the CLI extra arguments.

    These arguments will become the pipline input kwargs. The
    `parse_known_args` method doesn't handle `--key=value` and
    `--key value` automatically, so some care must be taken.

    Keep in mind that values are casted with json.
    """

    def parse(self, extras: list[str]) -> dict[str, Any]:
        """Parse extra arguments.

        Args:
            extras (list[str]): Extra arguments.

        Returns:
            dict[str, Any]: Parsed extra arguments.
        """
        extras = [arg for arg in extras if arg != "--"]
        kwargs: dict[str, Any] = {}
        nextras = len(extras)
        i = 0

        while i < nextras:
            arg = extras[i]
            if arg.startswith("-"):
                if "=" in arg:
                    raw_key, val_str = arg.split("=", 1)
                else:
                    raw_key = arg
                    if i + 1 < nextras and not extras[i + 1].startswith("-"):
                        val_str = extras[i + 1]
                        i += 1
                    else:
                        val_str = "true"

                key = raw_key.lstrip("-").replace("-", "_")
                kwargs[key] = self._maybe_cast_value(val_str)
            i += 1

        return kwargs

    def _maybe_cast_value(self, value: str) -> Any:
        """Try to cast the input value.

        In the future we might use the pipeline type hints.

        Args:
            value (str): Input value.

        Returns:
            Any: Parsed values. In case of parsing errors (e.g., for
                strings) the value is returned as-is.
        """
        try:
            return json.loads(value)
        except json.JSONDecodeError:
            return value

parse(extras)

Parse extra arguments.

Parameters:

Name	Type	Description	Default
extras	list[str]	Extra arguments.	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Parsed extra arguments.

Source code in python/sdag/parser.py

def parse(self, extras: list[str]) -> dict[str, Any]:
    """Parse extra arguments.

    Args:
        extras (list[str]): Extra arguments.

    Returns:
        dict[str, Any]: Parsed extra arguments.
    """
    extras = [arg for arg in extras if arg != "--"]
    kwargs: dict[str, Any] = {}
    nextras = len(extras)
    i = 0

    while i < nextras:
        arg = extras[i]
        if arg.startswith("-"):
            if "=" in arg:
                raw_key, val_str = arg.split("=", 1)
            else:
                raw_key = arg
                if i + 1 < nextras and not extras[i + 1].startswith("-"):
                    val_str = extras[i + 1]
                    i += 1
                else:
                    val_str = "true"

            key = raw_key.lstrip("-").replace("-", "_")
            kwargs[key] = self._maybe_cast_value(val_str)
        i += 1

    return kwargs

ParserBuilder

Build the parser.

Attributes:

Name	Type	Description
_parser	ArgumentParser	Main parser.
_subparsers	_SubParsersAction[ArgumentParser]	Subparsers for the different commands.

Source code in python/sdag/parser.py

class ParserBuilder:
    """Build the parser.

    Attributes:
        _parser (ArgumentParser): Main parser.
        _subparsers (_SubParsersAction[ArgumentParser]):
            Subparsers for the different commands.
    """

    def __init__(self):
        """Initialize the parser."""
        self._parser = ArgumentParser(description="SDAG")
        self._subparsers = self._parser.add_subparsers(
            help="Available commands"
        )

    def add_version(self) -> Self:
        """Add versions.

        Returns:
            Self: Parser builder.
        """
        self._parser.add_argument(
            "-v",
            "--version",
            action="version",
            version=f"%(prog)s {__version__}",
        )

        return self

    def add_compile_subparser(self) -> Self:
        """Add the compile command.

        Returns:
            Self: Parser builder.
        """
        compile_parser = self._subparsers.add_parser(
            "compile", help="Compile a pipeline."
        )

        compile_parser.add_argument(
            "pipeline",
            type=str,
            help=(
                "Pipeline name or import string (e.g., 'path.to:pipeline_fn')."
                " By default, pipelines are searched in './pipelines'. You can"
                " modify this behavior by setting the 'dag-dir' in the"
                " [tool.sdag] field of the pyproject.toml file."
            ),
        )

        self._add_compilation_args(compile_parser)
        self._add_log_level(compile_parser)
        self._add_detach(compile_parser)
        compile_parser.set_defaults(fn=compile_pipeline)

        return self

    def add_run_subparser(self) -> Self:
        """Add the run command.

        Returns:
            Self: Parser builder.
        """
        run_parser = self._subparsers.add_parser(
            "run", help="Compile and run a pipeline."
        )

        run_parser.add_argument(
            "pipeline",
            type=str,
            help=(
                "Pipeline name or import string (e.g., 'path.to:pipeline_fn')."
                " By default, pipelines are searched in './pipelines'. You can"
                " modify this behavior by setting the 'dag-dir' in the"
                " [tool.sdag] field of the pyproject.toml file. It can be"
                " equal to a valid compiled graph path to skip compilation."
            ),
        )

        self._add_log_level(run_parser)
        self._add_compilation_args(run_parser)
        self._add_scheduler_args(run_parser)
        self._add_local_option(run_parser)
        self._add_fail_fast_option(run_parser)
        self._add_detach(run_parser)
        run_parser.set_defaults(fn=run_pipeline)

        return self

    def add_restart_subparser(self) -> Self:
        """Add the restart subparser.

        Returns:
            Self: Parser.
        """
        restart_parser = self._subparsers.add_parser(
            "restart", help="Restart a previous run."
        )
        self._add_pipeline_name_and_hash_or_json(restart_parser)
        self._add_log_level(restart_parser)
        self._add_scheduler_args(restart_parser)
        self._add_fail_fast_option(restart_parser)
        self._add_detach(restart_parser)
        restart_parser.set_defaults(fn=restart_run)

        return self

    def add_retry_subparser(self) -> Self:
        """Add the retry subparser.

        Returns:
            Self: Parser.
        """
        restart_parser = self._subparsers.add_parser(
            "retry",
            help=(
                "Restart a previous run while resetting all"
                " failed and skipped tasks. If no tasks are failed"
                " or skipped, it is identical to sdag restart"
            ),
        )
        self._add_pipeline_name_and_hash_or_json(restart_parser)
        self._add_log_level(restart_parser)
        self._add_scheduler_args(restart_parser)
        self._add_fail_fast_option(restart_parser)
        self._add_detach(restart_parser)
        restart_parser.set_defaults(fn=retry_run)

        return self

    def add_kill_subparser(self) -> Self:
        """Add the pipeline kill command.

        Returns:
            Self: Parser builder.
        """
        kill_parser = self._subparsers.add_parser(
            "kill", help="Kill the runnig tasks of a pipeline."
        )

        self._add_pipeline_name_and_hash_or_json(kill_parser)
        self._add_log_level(kill_parser)
        self._add_detach(kill_parser)
        kill_parser.set_defaults(fn=kill_pipeline)

        return self

    def add_prune_subparser(self) -> Self:
        """Add the cache prune command.

        Returns:
            Self: Parser builder.
        """
        prune_parser = self._subparsers.add_parser(
            "prune", help="Prune the cache of a task."
        )
        prune_parser.add_argument(
            "task",
            nargs="*",
            type=str,
            help=(
                "Task to be pruned from the cache. Set to"
                " 'all' to prune the whole cache."
            ),
        )

        prune_parser.add_argument(
            "-p",
            "--pipeline",
            type=str,
            nargs="?",
            help=(
                "Pipeline. If the name is provided, the pipeline"
                " will be dynamically discovered. If the import string"
                " is given, autodiscovery will be skipped. If equal to"
                " a compiled JSON file, the pipeline data will be taken"
                " from there, thus skipping compilation entirely."
            ),
        )

        self._add_extra_metadata(prune_parser)
        self._add_log_level(prune_parser)
        self._add_detach(prune_parser)

        prune_parser.set_defaults(fn=prune_cache)

        return self

    def add_view_subparser(self) -> Self:
        """Add the view command.

        Returns:
            Self: Parser builder.
        """
        view_parser = self._subparsers.add_parser(
            "view", help="Visualize DAGs."
        )

        view_parser.add_argument(
            "pipeline",
            type=str,
            help=(
                "Pipeline. If the name is provided, the pipeline"
                " will be dynamically discovered. If the import string"
                " is given, autodiscovery will be skipped. If equal to"
                " a compiled JSON file, the pipeline data will be taken"
                " from there, thus skipping compilation entirely."
            ),
        )

        view_parser.add_argument(
            "--squeeze",
            action=BooleanOptionalAction,
            default=False,
            help="Collapse inner pipelines into a unique node",
        )

        self._add_extra_metadata(view_parser)
        self._add_log_level(view_parser)
        self._add_detach(view_parser)
        view_parser.set_defaults(fn=view_pipeline)

        return self

    def add_runtask_subparser(self) -> Self:
        """Add subparser to run tasks.

        Returns:
            Self: Parser.
        """
        runtask_parser = self._subparsers.add_parser(
            "runtask", help="Run a task."
        )

        runtask_parser.add_argument(
            "task",
            type=str,
            help="task executed",
        )

        runtask_parser.add_argument(
            "pipeline",
            type=str,
            help=(
                "Pipeline name or import string (e.g., 'path.to:pipeline_fn')."
            ),
        )

        self._add_log_level(runtask_parser)
        self._add_local_option(runtask_parser)
        self._add_fail_fast_option(runtask_parser)
        self._add_detach(runtask_parser)
        runtask_parser.set_defaults(fn=run_task)

        return self

    def add_list_subparser(self) -> Self:
        """Add Parser to list pipelines.

        Returns:
            Self: Parser.
        """
        list_parser = self._subparsers.add_parser(
            "list", help="List pipelines."
        )

        list_parser.add_argument(
            "pipeline",
            nargs="?",
            type=str,
            help=(
                "Pipeline name. If set, all the available"
                " runs for this pipeline will be shown."
            ),
        )
        self._add_log_level(list_parser)
        self._add_detach(list_parser)
        list_parser.set_defaults(fn=list_pipelines)

        return self

    def add_describe_subparser(self) -> Self:
        """Add the describe parser.

        Returns:
            Self: Parser.
        """
        describe_parser = self._subparsers.add_parser(
            "describe", help="Show input kwargs and caching info."
        )

        describe_parser.add_argument(
            "pipeline",
            type=str,
            help=(
                "Pipeline name or import string (e.g., 'path.to:pipeline_fn')."
                " By default, pipelines are searched in './pipelines'. You can"
                " modify this behavior by setting the 'dag-dir' in the"
                " [tool.sdag] field of the pyproject.toml file. It can be"
                " equal to a valid compiled graph path to skip compilation."
            ),
        )

        self._add_compilation_args(describe_parser)
        self._add_log_level(describe_parser)
        self._add_detach(describe_parser)
        describe_parser.set_defaults(fn=describe_pipeline)

        return self

    def add_status_subparser(self) -> Self:
        """Add the status parser.

        Returns:
            Self: Parser.
        """
        status_parser = self._subparsers.add_parser(
            "status",
            help=(
                "Read the last checkpoint of the scheduler for"
                "a pipeline and print the status table."
            ),
        )
        self._add_pipeline_name_and_hash_or_json(status_parser)
        self._add_log_level(status_parser)
        self._add_detach(status_parser)
        status_parser.set_defaults(fn=print_status)

        return self

    def get_parser(self) -> ArgumentParser:
        """Get the parser.

        Returns:
            ArgumentParser: Parser.
        """
        return self._parser

    def _add_detach(self, parser: ArgumentParser) -> None:
        """Add the detach option to the parser.

        Detached commands run on Slurm.

        Args:
            parser (ArgumentParser): Parser.
        """
        parser.add_argument(
            "--detach",
            action=BooleanOptionalAction,
            default=False,
            help="Run the command as a sbatch job",
        )

    def _add_log_level(self, subparser: ArgumentParser) -> Self:
        """Add the log level.

        Args:
            subparser (ArgumentParser): Subparser.

        Returns:
            Self: Parser builder.
        """
        subparser.add_argument(
            "-l",
            "--log-level",
            choices=["debug", "info", "warning", "error"],
            default=None,
            help="Scheduler log level. Defaults to 'info'",
        )

        return self

    def _add_pipeline_name_and_hash_or_json(
        self, subparser: ArgumentParser
    ) -> None:
        """Add the pipeline name and hash.

        Args:
            subparser (ArgumentParser): Subparser.
        """
        subparser.add_argument(
            "pipeline",
            type=str,
            help="Pipeline name.",
        )

        subparser.add_argument(
            "--hash",
            type=str,
            default="last",
            help=(
                "Pipeline hash as written in the compiled JSON file."
                " If equal to 'last', the most recently created pipeline"
                " is considered. Defaults to 'last'"
            ),
        )

    def _add_scheduler_args(self, subparser: ArgumentParser) -> None:
        """Add the scheduler options.

        Args:
            subparser (ArgumentParser): Subparser.
        """
        subparser.add_argument(
            "-t",
            "--time-between-polls",
            type=int,
            default=None,
            help="Time between slurm polls (seconds). Defaults to 5s.",
        )

        subparser.add_argument(
            "-m",
            "--max-concurrency",
            type=int,
            default=None,
            help=(
                "Maximum number of jobs running at the same time."
                "Set to 0 for unlimited concurrency. Defaults to 10."
            ),
        )

    def _add_local_option(self, subparser: ArgumentParser) -> None:
        """Add local flag.

        Args:
            subparser (ArgumentParser): Parser.
        """
        subparser.add_argument(
            "--local",
            action=BooleanOptionalAction,
            default=False,
            help="Run tasks locally as nonblocking child processes.",
        )

    def _add_fail_fast_option(self, subparser: ArgumentParser) -> None:
        """Add the fail fast flag.

        Args:
            subparser (ArgumentParser): Parser.
        """
        subparser.add_argument(
            "--fail-fast",
            action=BooleanOptionalAction,
            default=False,
            help=(
                "Kill the scheduler and all running tasks if any of them fails"
            ),
        )

    def _add_compilation_args(self, subparser: ArgumentParser) -> None:
        """Add the compilation options.

        Args:
            subparser (ArgumentParser): Subparser.
        """
        subparser.add_argument(
            "-d",
            "--dst-dir",
            type=str,
            default=None,
            help="Destination directory. Defaults to the current one.",
        )

        subparser.add_argument(
            "-n",
            "--name",
            type=str,
            default=None,
            help="Compiled DAG file name. Defaults to <pipeline-name>.json.",
        )

        self._add_extra_metadata(subparser)

    def _add_extra_metadata(self, subparser: ArgumentParser) -> None:
        """Add extra metadata for the pipeline as a JSON string.

        Args:
            subparser (ArgumentParser): Subparser.
        """
        subparser.add_argument(
            "-e",
            "--extra-metadata",
            type=str,
            default=None,
            help="Extra metadata as a JSON string. Defaults to None",
        )

__init__()

Initialize the parser.

Source code in python/sdag/parser.py

def __init__(self):
    """Initialize the parser."""
    self._parser = ArgumentParser(description="SDAG")
    self._subparsers = self._parser.add_subparsers(
        help="Available commands"
    )

add_compile_subparser()

Add the compile command.

Returns:

Name	Type	Description
Self	Self	Parser builder.

Source code in python/sdag/parser.py

def add_compile_subparser(self) -> Self:
    """Add the compile command.

    Returns:
        Self: Parser builder.
    """
    compile_parser = self._subparsers.add_parser(
        "compile", help="Compile a pipeline."
    )

    compile_parser.add_argument(
        "pipeline",
        type=str,
        help=(
            "Pipeline name or import string (e.g., 'path.to:pipeline_fn')."
            " By default, pipelines are searched in './pipelines'. You can"
            " modify this behavior by setting the 'dag-dir' in the"
            " [tool.sdag] field of the pyproject.toml file."
        ),
    )

    self._add_compilation_args(compile_parser)
    self._add_log_level(compile_parser)
    self._add_detach(compile_parser)
    compile_parser.set_defaults(fn=compile_pipeline)

    return self

add_describe_subparser()

Add the describe parser.

Returns:

Name	Type	Description
Self	Self	Parser.

Source code in python/sdag/parser.py

def add_describe_subparser(self) -> Self:
    """Add the describe parser.

    Returns:
        Self: Parser.
    """
    describe_parser = self._subparsers.add_parser(
        "describe", help="Show input kwargs and caching info."
    )

    describe_parser.add_argument(
        "pipeline",
        type=str,
        help=(
            "Pipeline name or import string (e.g., 'path.to:pipeline_fn')."
            " By default, pipelines are searched in './pipelines'. You can"
            " modify this behavior by setting the 'dag-dir' in the"
            " [tool.sdag] field of the pyproject.toml file. It can be"
            " equal to a valid compiled graph path to skip compilation."
        ),
    )

    self._add_compilation_args(describe_parser)
    self._add_log_level(describe_parser)
    self._add_detach(describe_parser)
    describe_parser.set_defaults(fn=describe_pipeline)

    return self

add_kill_subparser()

Add the pipeline kill command.

Returns:

Name	Type	Description
Self	Self	Parser builder.

Source code in python/sdag/parser.py

def add_kill_subparser(self) -> Self:
    """Add the pipeline kill command.

    Returns:
        Self: Parser builder.
    """
    kill_parser = self._subparsers.add_parser(
        "kill", help="Kill the runnig tasks of a pipeline."
    )

    self._add_pipeline_name_and_hash_or_json(kill_parser)
    self._add_log_level(kill_parser)
    self._add_detach(kill_parser)
    kill_parser.set_defaults(fn=kill_pipeline)

    return self

add_list_subparser()

Add Parser to list pipelines.

Returns:

Name	Type	Description
Self	Self	Parser.

Source code in python/sdag/parser.py

def add_list_subparser(self) -> Self:
    """Add Parser to list pipelines.

    Returns:
        Self: Parser.
    """
    list_parser = self._subparsers.add_parser(
        "list", help="List pipelines."
    )

    list_parser.add_argument(
        "pipeline",
        nargs="?",
        type=str,
        help=(
            "Pipeline name. If set, all the available"
            " runs for this pipeline will be shown."
        ),
    )
    self._add_log_level(list_parser)
    self._add_detach(list_parser)
    list_parser.set_defaults(fn=list_pipelines)

    return self

add_prune_subparser()

Add the cache prune command.

Returns:

Name	Type	Description
Self	Self	Parser builder.

Source code in python/sdag/parser.py

def add_prune_subparser(self) -> Self:
    """Add the cache prune command.

    Returns:
        Self: Parser builder.
    """
    prune_parser = self._subparsers.add_parser(
        "prune", help="Prune the cache of a task."
    )
    prune_parser.add_argument(
        "task",
        nargs="*",
        type=str,
        help=(
            "Task to be pruned from the cache. Set to"
            " 'all' to prune the whole cache."
        ),
    )

    prune_parser.add_argument(
        "-p",
        "--pipeline",
        type=str,
        nargs="?",
        help=(
            "Pipeline. If the name is provided, the pipeline"
            " will be dynamically discovered. If the import string"
            " is given, autodiscovery will be skipped. If equal to"
            " a compiled JSON file, the pipeline data will be taken"
            " from there, thus skipping compilation entirely."
        ),
    )

    self._add_extra_metadata(prune_parser)
    self._add_log_level(prune_parser)
    self._add_detach(prune_parser)

    prune_parser.set_defaults(fn=prune_cache)

    return self

add_restart_subparser()

Add the restart subparser.

Returns:

Name	Type	Description
Self	Self	Parser.

Source code in python/sdag/parser.py

def add_restart_subparser(self) -> Self:
    """Add the restart subparser.

    Returns:
        Self: Parser.
    """
    restart_parser = self._subparsers.add_parser(
        "restart", help="Restart a previous run."
    )
    self._add_pipeline_name_and_hash_or_json(restart_parser)
    self._add_log_level(restart_parser)
    self._add_scheduler_args(restart_parser)
    self._add_fail_fast_option(restart_parser)
    self._add_detach(restart_parser)
    restart_parser.set_defaults(fn=restart_run)

    return self

add_retry_subparser()

Add the retry subparser.

Returns:

Name	Type	Description
Self	Self	Parser.

Source code in python/sdag/parser.py

def add_retry_subparser(self) -> Self:
    """Add the retry subparser.

    Returns:
        Self: Parser.
    """
    restart_parser = self._subparsers.add_parser(
        "retry",
        help=(
            "Restart a previous run while resetting all"
            " failed and skipped tasks. If no tasks are failed"
            " or skipped, it is identical to sdag restart"
        ),
    )
    self._add_pipeline_name_and_hash_or_json(restart_parser)
    self._add_log_level(restart_parser)
    self._add_scheduler_args(restart_parser)
    self._add_fail_fast_option(restart_parser)
    self._add_detach(restart_parser)
    restart_parser.set_defaults(fn=retry_run)

    return self

add_run_subparser()

Add the run command.

Returns:

Name	Type	Description
Self	Self	Parser builder.

Source code in python/sdag/parser.py

def add_run_subparser(self) -> Self:
    """Add the run command.

    Returns:
        Self: Parser builder.
    """
    run_parser = self._subparsers.add_parser(
        "run", help="Compile and run a pipeline."
    )

    run_parser.add_argument(
        "pipeline",
        type=str,
        help=(
            "Pipeline name or import string (e.g., 'path.to:pipeline_fn')."
            " By default, pipelines are searched in './pipelines'. You can"
            " modify this behavior by setting the 'dag-dir' in the"
            " [tool.sdag] field of the pyproject.toml file. It can be"
            " equal to a valid compiled graph path to skip compilation."
        ),
    )

    self._add_log_level(run_parser)
    self._add_compilation_args(run_parser)
    self._add_scheduler_args(run_parser)
    self._add_local_option(run_parser)
    self._add_fail_fast_option(run_parser)
    self._add_detach(run_parser)
    run_parser.set_defaults(fn=run_pipeline)

    return self

add_runtask_subparser()

Add subparser to run tasks.

Returns:

Name	Type	Description
Self	Self	Parser.

Source code in python/sdag/parser.py

def add_runtask_subparser(self) -> Self:
    """Add subparser to run tasks.

    Returns:
        Self: Parser.
    """
    runtask_parser = self._subparsers.add_parser(
        "runtask", help="Run a task."
    )

    runtask_parser.add_argument(
        "task",
        type=str,
        help="task executed",
    )

    runtask_parser.add_argument(
        "pipeline",
        type=str,
        help=(
            "Pipeline name or import string (e.g., 'path.to:pipeline_fn')."
        ),
    )

    self._add_log_level(runtask_parser)
    self._add_local_option(runtask_parser)
    self._add_fail_fast_option(runtask_parser)
    self._add_detach(runtask_parser)
    runtask_parser.set_defaults(fn=run_task)

    return self

add_status_subparser()

Add the status parser.

Returns:

Name	Type	Description
Self	Self	Parser.

Source code in python/sdag/parser.py

def add_status_subparser(self) -> Self:
    """Add the status parser.

    Returns:
        Self: Parser.
    """
    status_parser = self._subparsers.add_parser(
        "status",
        help=(
            "Read the last checkpoint of the scheduler for"
            "a pipeline and print the status table."
        ),
    )
    self._add_pipeline_name_and_hash_or_json(status_parser)
    self._add_log_level(status_parser)
    self._add_detach(status_parser)
    status_parser.set_defaults(fn=print_status)

    return self

add_version()

Add versions.

Returns:

Name	Type	Description
Self	Self	Parser builder.

Source code in python/sdag/parser.py

def add_version(self) -> Self:
    """Add versions.

    Returns:
        Self: Parser builder.
    """
    self._parser.add_argument(
        "-v",
        "--version",
        action="version",
        version=f"%(prog)s {__version__}",
    )

    return self

add_view_subparser()

Add the view command.

Returns:

Name	Type	Description
Self	Self	Parser builder.

Source code in python/sdag/parser.py

def add_view_subparser(self) -> Self:
    """Add the view command.

    Returns:
        Self: Parser builder.
    """
    view_parser = self._subparsers.add_parser(
        "view", help="Visualize DAGs."
    )

    view_parser.add_argument(
        "pipeline",
        type=str,
        help=(
            "Pipeline. If the name is provided, the pipeline"
            " will be dynamically discovered. If the import string"
            " is given, autodiscovery will be skipped. If equal to"
            " a compiled JSON file, the pipeline data will be taken"
            " from there, thus skipping compilation entirely."
        ),
    )

    view_parser.add_argument(
        "--squeeze",
        action=BooleanOptionalAction,
        default=False,
        help="Collapse inner pipelines into a unique node",
    )

    self._add_extra_metadata(view_parser)
    self._add_log_level(view_parser)
    self._add_detach(view_parser)
    view_parser.set_defaults(fn=view_pipeline)

    return self

get_parser()

Get the parser.

Returns:

Name	Type	Description
ArgumentParser	ArgumentParser	Parser.

Source code in python/sdag/parser.py

def get_parser(self) -> ArgumentParser:
    """Get the parser.

    Returns:
        ArgumentParser: Parser.
    """
    return self._parser

pyproj

Settings.

DetachOptions

Bases: SlurmOverride

Detached sdag command options.

Attributes:

Name	Type	Description
job_name	str | None	Slurm job name. Defaults to 'sdag'.
nodes	str | None	Number of nodes. Defaults to 1.
ntasks_per_node	str | None	Number of tasks per node. Defaults to 1.
cpus_per_task	str | None	Number of CPUs per task. Defaults to 1.
output	str | None	Slurm stdout. Defaults to 'logs/sdag/sdag_%x.%j.out'.
error	str | None	Slurm stderr. Defaults to 'logs/sdag/sdag_%x.%j.err'.

Source code in python/sdag/pyproj.py

class DetachOptions(SlurmOverride):
    """Detached sdag command options.

    Attributes:
        job_name (str | None): Slurm job name. Defaults to 'sdag'.
        nodes (str | None): Number of nodes. Defaults to 1.
        ntasks_per_node (str | None): Number of tasks per node.
            Defaults to 1.
        cpus_per_task (str | None): Number of CPUs per task.
            Defaults to 1.
        output (str | None): Slurm stdout. Defaults to
            'logs/sdag/sdag_%x.%j.out'.
        error (str | None): Slurm stderr. Defaults to
            'logs/sdag/sdag_%x.%j.err'.
    """

    job_name: str | None = Field(alias="job-name", default="sdag")
    nodes: NonNegativeInt | None = 1
    ntasks_per_node: NonNegativeInt | None = Field(
        alias="ntasks-per-node", default=1
    )
    cpus_per_task: NonNegativeInt | None = Field(
        alias="cpus-per-task", default=1
    )
    output: str | None = r"logs/sdag/sdag_%x.%j.out"
    error: str | None = r"logs/sdag/sdag_%x.%j.err"

    def get_detached_args(self) -> list[str]:
        """Get the slurm detach args.

        Returns:
            list[str]: Args used by sbatch to submit the
                detached job.
        """
        args: list[str] = []
        dumped = self.model_dump(exclude_none=True)
        for key, value in dumped.items():
            args.append(f"--{key}")
            args.append(f"{value}")
        return args

get_detached_args()

Get the slurm detach args.

Returns:

Type	Description
list[str]	list[str]: Args used by sbatch to submit the detached job.

Source code in python/sdag/pyproj.py

def get_detached_args(self) -> list[str]:
    """Get the slurm detach args.

    Returns:
        list[str]: Args used by sbatch to submit the
            detached job.
    """
    args: list[str] = []
    dumped = self.model_dump(exclude_none=True)
    for key, value in dumped.items():
        args.append(f"--{key}")
        args.append(f"{value}")
    return args

Pyproj

Bases: SlurmOverride

pyproject.toml configs.

Attributes:

Name	Type	Description
dag_dir	str	DAG directory.
compiled_dag_dir	str	Compiled DAG directory.
prepend_compiled_dag_dir	bool	Add the compiled pipeline dir automatically if the directory has not been specified.
time_between_polls	int	Time between successive Slurm polls in seconds. Defaults to 5.
commands	Commands | None	Commands to execute tasks. If set, it overrides all pipeline commands.
log_level	Literal['debug', 'info', 'warning', 'error']	Logging level.
max_concurrency	int	Maximum number of tasks executed at the same time.
slurm_grace_period	int	Number of turns a task can be missing from the response without being considered as failed. Defaults to 60.
detach	DetachOptions	Detached command options. They are ignored if the command is not detached.
fail_fast	bool	Kill the scheduler and all running tasks if any of them fails.
envs	dict[str, str]	Environment variables added to all tasks. They have higher priority with respect to the ones set inside pipelines.

Source code in python/sdag/pyproj.py

class Pyproj(SlurmOverride):
    """pyproject.toml configs.

    Attributes:
        dag_dir (str): DAG directory.
        compiled_dag_dir (str): Compiled DAG directory.
        prepend_compiled_dag_dir (bool): Add the compiled pipeline
            dir automatically if the directory has not been
            specified.
        time_between_polls (int): Time between successive Slurm polls
            in seconds. Defaults to 5.
        commands (Commands | None): Commands to execute tasks. If
            set, it overrides all pipeline commands.
        log_level (Literal["debug", "info", "warning", "error"]):
            Logging level.
        max_concurrency (int): Maximum number of tasks executed
            at the same time.
        slurm_grace_period (int): Number of turns a task can be missing
            from the response without being considered as failed.
            Defaults to 60.
        detach (DetachOptions): Detached command options. They are
            ignored if the command is not detached.
        fail_fast (bool): Kill the scheduler and all running tasks if
            any of them fails.
        envs (dict[str, str]): Environment variables added to all tasks.
            They have higher priority with respect to the ones set inside
            pipelines.
    """

    dag_dir: str = Field(alias="dag-dir", default=".")
    compiled_dag_dir: str = Field(
        alias="compiled-dag-dir", default="./compiled-pipelines"
    )
    cmd: Commands | None = None
    prepend_compiled_dag_dir: bool = Field(
        alias="prepend-compiled-dag-dir", default=True
    )
    log_level: Literal["debug", "info", "warning", "error"] = Field(
        alias="log-level", default="info"
    )
    time_between_polls: int = Field(
        alias="time-between-polls", default=5, ge=0
    )
    max_concurrency: int = Field(alias="max-concurrency", default=0, ge=0)
    slurm_grace_period: int = Field(
        alias="slurm-grace-period", default=60, ge=0
    )
    max_concurrent_runs: int = Field(
        alias="max-concurrent-runs", default=20, gt=0
    )
    fail_fast: bool = Field(alias="fail-fast", default=False)
    detach: DetachOptions = Field(default_factory=DetachOptions)
    envs: dict[str, str] = Field(default_factory=dict)
    tags: list[SDAGTag] = Field(default_factory=list)

    model_config = ConfigDict(
        extra="forbid", validate_by_alias=True, serialize_by_alias=True
    )

    @model_validator(mode="after")
    def join_args_into_detach(self) -> Self:
        """Override detach args with the pyproj ones.

        account, partition, qos, and time are overwritten with the
        default values in the pyproj.

        Returns:
            Self: pyproj.
        """
        if self.detach.account is None:
            self.detach.account = self.account

        if self.detach.partition is None:
            self.detach.partition = self.partition

        if self.detach.qos is None:
            self.detach.qos = self.qos

        if self.detach.time is None:
            self.detach.time = self.time

        return self

    def join_cli_args(self, args: Namespace) -> None:
        """Join the user preferences into the pyproject.

        Args:
            args (Namespace): Parsed CLI args.
        """
        argdict = args.__dict__

        if argdict.get("log_level") is not None:
            self.log_level = args.log_level

        if argdict.get("time_between_polls") is not None:
            self.time_between_polls = args.time_between_polls

        if argdict.get("max_concurrency") is not None:
            self.max_concurrency = args.max_concurrency

        if argdict.get("compiled_dag_dir") is not None:
            self.compiled_dag_dir = args.compiled_dag_dir

        if argdict.get("fail_fast"):
            self.fail_fast = True

        if argdict.get("local"):
            self.cmd = "bash"
            for tag in self.tags:
                tag.cmd = "bash"

join_args_into_detach()

Override detach args with the pyproj ones.

account, partition, qos, and time are overwritten with the default values in the pyproj.

Returns:

Name	Type	Description
Self	Self	pyproj.

Source code in python/sdag/pyproj.py

@model_validator(mode="after")
def join_args_into_detach(self) -> Self:
    """Override detach args with the pyproj ones.

    account, partition, qos, and time are overwritten with the
    default values in the pyproj.

    Returns:
        Self: pyproj.
    """
    if self.detach.account is None:
        self.detach.account = self.account

    if self.detach.partition is None:
        self.detach.partition = self.partition

    if self.detach.qos is None:
        self.detach.qos = self.qos

    if self.detach.time is None:
        self.detach.time = self.time

    return self

join_cli_args(args)

Join the user preferences into the pyproject.

Parameters:

Name	Type	Description	Default
args	Namespace	Parsed CLI args.	required

Source code in python/sdag/pyproj.py

def join_cli_args(self, args: Namespace) -> None:
    """Join the user preferences into the pyproject.

    Args:
        args (Namespace): Parsed CLI args.
    """
    argdict = args.__dict__

    if argdict.get("log_level") is not None:
        self.log_level = args.log_level

    if argdict.get("time_between_polls") is not None:
        self.time_between_polls = args.time_between_polls

    if argdict.get("max_concurrency") is not None:
        self.max_concurrency = args.max_concurrency

    if argdict.get("compiled_dag_dir") is not None:
        self.compiled_dag_dir = args.compiled_dag_dir

    if argdict.get("fail_fast"):
        self.fail_fast = True

    if argdict.get("local"):
        self.cmd = "bash"
        for tag in self.tags:
            tag.cmd = "bash"

SDAGTag

Bases: SlurmOverride

Tag configurations.

User are allowed to override slurm variables, command, and script path.

Attributes:

Name	Type	Description
tag	str	Tag name.
cmd	Commands | None	Command. Defaults to None.
script_path	Path | None	Script path.
envs	dict[str, str]	Environment variables added to all relevant tasks.

Source code in python/sdag/pyproj.py

class SDAGTag(SlurmOverride):
    """Tag configurations.

    User are allowed to override slurm variables,
    command, and script path.

    Attributes:
        tag (str): Tag name.
        cmd (Commands | None): Command. Defaults to None.
        script_path (Path | None): Script path.
        envs (dict[str, str]): Environment variables
            added to all relevant tasks.
    """

    tag: str
    cmd: Commands | None = None
    script_path: Path | None = None
    envs: dict[str, str] = Field(default_factory=dict)

apply_pyproj_configs(dag, pyproj)

Apply pyproject settings to all tasks.

Parameters:

Name	Type	Description	Default
dag	DAG	Compiled DAG.	required
pyproj	Pyproj	Parsed pyproject.toml.	required

Source code in python/sdag/pyproj.py

def apply_pyproj_configs(dag: DAG, pyproj: Pyproj) -> None:
    """Apply pyproject settings to all tasks.

    Args:
        dag (DAG): Compiled DAG.
        pyproj (Pyproj): Parsed pyproject.toml.
    """
    for task in dag.nodes:
        if task.kind == "task":
            apply_pyproj_to_task(task, pyproj)

apply_pyproj_to_task(task, pyproj)

Apply pyproject settings to a task.

Parameters:

Name	Type	Description	Default
task	TaskNode	Task node.	required
pyproj	Pyproj	Parsed pyproject.	required

Source code in python/sdag/pyproj.py

def apply_pyproj_to_task(task: TaskNode, pyproj: Pyproj) -> None:
    """Apply pyproject settings to a task.

    Args:
        task (TaskNode): Task node.
        pyproj (Pyproj): Parsed pyproject.
    """
    if pyproj.cmd is not None:
        task.cmd = pyproj.cmd
    task.slurm.override_from(pyproj)
    task.envs |= pyproj.envs
    for tag in pyproj.tags:
        if tag.tag in task.tags:
            _apply_tag_configs(task, tag)

get_pyproj(pyproj_path='pyproject.toml', sdag_toml_path='.sdag.toml') cached

Get configurations.

Parameters:

Name	Type	Description	Default
pyproj_path	str	pyproject.toml path. Defaults to "pyproject.toml".	'pyproject.toml'
sdag_toml_path	str	sdag.toml path. Defaults to ".sdag.toml".	'.sdag.toml'

Returns:

Name	Type	Description
Pyproj	Pyproj	Parsed configurations. User settings have not been added.

Source code in python/sdag/pyproj.py

@lru_cache(maxsize=1)
def get_pyproj(
    pyproj_path: str = "pyproject.toml", sdag_toml_path: str = ".sdag.toml"
) -> Pyproj:
    """Get configurations.

    Args:
        pyproj_path (str, optional): pyproject.toml path. Defaults
            to "pyproject.toml".
        sdag_toml_path (str, optional): sdag.toml path. Defaults
            to ".sdag.toml".

    Returns:
        Pyproj: Parsed configurations. User settings have not been
            added.
    """
    pyproj_dict = _load_toml(pyproj_path).get("tool", {}).get("sdag", {})
    sdag_toml_dict = _load_toml(sdag_toml_path)
    joined_dict = _join_configs(pyproj_dict, sdag_toml_dict)

    return Pyproj.model_validate(joined_dict, by_alias=True)

settings

Settings.

CompileSettings

Bases: BaseSettings

Compilation environment variables.

Attributes:

Name	Type	Description
sdag_base_path	Path | None	Base path. Defaults to None.

Source code in python/sdag/settings.py

class CompileSettings(BaseSettings):
    """Compilation environment variables.

    Attributes:
        sdag_base_path (Path | None): Base path. Defaults to None.
    """

    sdag_base_path: Path | None = None

EnvLoggerFormatter

Bases: Formatter

Formatter mimicking env_logger.

Attributes:

Name	Type	Description
_crop	dict[str, str]	Used to crop log levels as env_logger does.
_colors	dict[str, str]	Logging level colors.

Source code in python/sdag/settings.py

class EnvLoggerFormatter(logging.Formatter):
    """Formatter mimicking env_logger.

    Attributes:
        _crop (dict[str, str]): Used to crop log levels as
            env_logger does.
        _colors (dict[str, str]): Logging level colors.
    """

    def __init__(self):
        """Initialize the log formatter."""
        self._crop = {"WARNING": "WARN", "CRITICAL": "CRIT"}
        self._colors = {
            "DEBUG": LogColor.BLUE,
            "INFO": LogColor.GREEN,
            "WARN": LogColor.YELLOW,
            "ERROR": LogColor.RED,
            "CRIT": LogColor.RED,
        }
        fmt = "[%(asctime)s %(levelname)-5s %(name)s] %(message)s"
        super().__init__(fmt=fmt, datefmt="%Y-%m-%dT%H:%M:%SZ")

    def format(self, record: logging.LogRecord) -> str:
        """Format a record.

        Args:
            record (logging.LogRecord): Raw record.

        Returns:
            str: Formatted record.
        """
        record.levelname = self._format_levelname(record.levelname)
        return super().format(record)

    def _format_levelname(self, levelname: str) -> str:
        """Format the level name like env_logger does.

        The level name is cropped or padded to 5 characters
        and colored.

        Args:
            levelname (str): Levelname.

        Returns:
            str: Formatted levelname.
        """
        name = self._crop.get(levelname, levelname)
        color = self._colors.get(name, "INFO")
        return f"{color}{name:<5}{LogColor.RESET}"

__init__()

Initialize the log formatter.

Source code in python/sdag/settings.py

def __init__(self):
    """Initialize the log formatter."""
    self._crop = {"WARNING": "WARN", "CRITICAL": "CRIT"}
    self._colors = {
        "DEBUG": LogColor.BLUE,
        "INFO": LogColor.GREEN,
        "WARN": LogColor.YELLOW,
        "ERROR": LogColor.RED,
        "CRIT": LogColor.RED,
    }
    fmt = "[%(asctime)s %(levelname)-5s %(name)s] %(message)s"
    super().__init__(fmt=fmt, datefmt="%Y-%m-%dT%H:%M:%SZ")

format(record)

Format a record.

Parameters:

Name	Type	Description	Default
record	LogRecord	Raw record.	required

Returns:

Name	Type	Description
str	str	Formatted record.

Source code in python/sdag/settings.py

def format(self, record: logging.LogRecord) -> str:
    """Format a record.

    Args:
        record (logging.LogRecord): Raw record.

    Returns:
        str: Formatted record.
    """
    record.levelname = self._format_levelname(record.levelname)
    return super().format(record)

LogColor

Bases: StrEnum

Logging colors matching with env_logger.

Source code in python/sdag/settings.py

class LogColor(StrEnum):
    """Logging colors matching with env_logger."""

    BLUE = "\033[34m"
    GREEN = "\033[32m"
    YELLOW = "\033[33m"
    RED = "\033[31m"
    RESET = "\033[0m"

Settings

Bases: BaseSettings

Env variables set by the scheduler.

Attributes:

Name	Type	Description
sdag_pipeline_dir	str	Pipeline directory.
sdag_pipeline_name	str	Pipeline name.
sdag_subpipeline_name	str	Task pipeline name. If there are no nested pipeline, it is equal to the pipeline name.
sdag_import_path	str	Import path used to compile the pipeline.
sdag_input_kwargs	str	Pipeline input kwargs JSON string. Useful to recompile the pipeline.
sdag_uid	str	Node unique id.
sdag_task_fn	str	Task function name.
sdag_task_name	str	Task name, by default it's equal to the task name.
sdag_each_key	str | None	If set, it's the sharded key in job arrays.
sdag_shard_path	str | None	If set, it's the shard path in job arrays.
slurm_array_task_id	str | None	If set, it's the task id in job arrays.
sdag_try_num	int	Number of times the task has been executed.

Source code in python/sdag/settings.py

class Settings(BaseSettings):
    """Env variables set by the scheduler.

    Attributes:
        sdag_pipeline_dir (str): Pipeline directory.
        sdag_pipeline_name (str): Pipeline name.
        sdag_subpipeline_name (str): Task pipeline name.
            If there are no nested pipeline, it is equal
            to the pipeline name.
        sdag_import_path (str): Import path used to compile
            the pipeline.
        sdag_input_kwargs (str): Pipeline input kwargs
            JSON string. Useful to recompile the pipeline.
        sdag_uid (str): Node unique id.
        sdag_task_fn (str): Task function name.
        sdag_task_name (str): Task name, by default it's equal to
            the task name.
        sdag_each_key (str | None): If set, it's the sharded key in
            job arrays.
        sdag_shard_path (str | None): If set, it's the shard path in
            job arrays.
        slurm_array_task_id (str | None): If set, it's the task id
            in job arrays.
        sdag_try_num (int): Number of times the task has been executed.
    """

    sdag_pipeline_dir: Path
    sdag_pipeline_name: str
    sdag_subpipeline_name: str
    sdag_import_path: str
    sdag_input_kwargs: str
    sdag_uid: int
    sdag_task_name: str
    sdag_task_fn: str
    sdag_try_num: int
    sdag_log_level: str
    sdag_shard_path: Path | None = None
    sdag_each_key: str | None = None
    slurm_array_task_id: int | None = None

    @model_validator(mode="after")
    def shard_paths_and_task_id_must_be_consistent(self) -> Self:
        """Both must be either set or null.

        Returns:
            Self: Settings.
        """
        shard_path, task_id, each = (
            self.sdag_shard_path,
            self.slurm_array_task_id,
            self.sdag_each_key,
        )
        shard_ok = shard_path is not None and bool(shard_path)
        task_id_ok = task_id is not None
        each_ok = each is not None and bool(each)
        if shard_ok != task_id_ok or task_id_ok != each_ok:
            raise InconsistentArraySettingsError(shard_path, task_id, each)

        return self

    def is_job_array(self) -> bool:
        """Check if the job is part of an array.

        Returns:
            bool: True if the job is part of a job array.
        """
        shard_path, task_id, each = (
            self.sdag_shard_path,
            self.slurm_array_task_id,
            self.sdag_each_key,
        )
        shard_ok = shard_path is not None and bool(shard_path)
        task_id_ok = task_id is not None
        each_ok = each is not None and bool(each)

        return shard_ok and task_id_ok and each_ok

is_job_array()

Check if the job is part of an array.

Returns:

Name	Type	Description
bool	bool	True if the job is part of a job array.

Source code in python/sdag/settings.py

def is_job_array(self) -> bool:
    """Check if the job is part of an array.

    Returns:
        bool: True if the job is part of a job array.
    """
    shard_path, task_id, each = (
        self.sdag_shard_path,
        self.slurm_array_task_id,
        self.sdag_each_key,
    )
    shard_ok = shard_path is not None and bool(shard_path)
    task_id_ok = task_id is not None
    each_ok = each is not None and bool(each)

    return shard_ok and task_id_ok and each_ok

shard_paths_and_task_id_must_be_consistent()

Both must be either set or null.

Returns:

Name	Type	Description
Self	Self	Settings.

Source code in python/sdag/settings.py

@model_validator(mode="after")
def shard_paths_and_task_id_must_be_consistent(self) -> Self:
    """Both must be either set or null.

    Returns:
        Self: Settings.
    """
    shard_path, task_id, each = (
        self.sdag_shard_path,
        self.slurm_array_task_id,
        self.sdag_each_key,
    )
    shard_ok = shard_path is not None and bool(shard_path)
    task_id_ok = task_id is not None
    each_ok = each is not None and bool(each)
    if shard_ok != task_id_ok or task_id_ok != each_ok:
        raise InconsistentArraySettingsError(shard_path, task_id, each)

    return self

configure_logging(log_level)

Simple logging configuration.

Everything is streamed to the stderr.

Parameters:

Name	Type	Description	Default
log_level	str	Logging level.	required

Source code in python/sdag/settings.py

def configure_logging(log_level: str) -> None:
    """Simple logging configuration.

    Everything is streamed to the stderr.

    Args:
        log_level (str): Logging level.
    """
    handler = logging.StreamHandler()
    handler.setFormatter(EnvLoggerFormatter())
    logging.basicConfig(level=log_level.upper(), handlers=[handler])

get_cached_settings() cached

Get the SDAG environment variables.

Returns:

Name	Type	Description
Settings	Settings	Parsed environment variables.

Source code in python/sdag/settings.py

@lru_cache
def get_cached_settings() -> Settings:
    """Get the SDAG environment variables.

    Returns:
        Settings: Parsed environment variables.
    """
    return Settings()  # type: ignore

get_compile_settings() cached

Get cached compilation settings.

Returns:

Name	Type	Description
CompileSettings	CompileSettings	Cached compilation settings.

Source code in python/sdag/settings.py

@lru_cache
def get_compile_settings() -> CompileSettings:
    """Get cached compilation settings.

    Returns:
        CompileSettings: Cached compilation settings.
    """
    return CompileSettings()

types

Types.

CacheOptions = Literal['none', 'output', 'io', 'task', 'project'] module-attribute

Cache options.

Commands = Literal['bash', 'sbatch'] module-attribute

Available commands.

ExecMode = Literal['wrap', 'ext'] module-attribute

Available execution modes.

Scope = Literal['local', 'global'] module-attribute

Task scopes.

visualization

Visualization utils.

For notebook visualizations it is required to install neo4j_viz.

DAGViewer

DAG viewer.

Attributes:

Name	Type	Description
squeeze	bool	Collapse all inner pipelines into a unique node.
task_size	int	Task node size.
task_color	str	Task color.
root_size	int	Root node size.
root_color	str	Root color.
end_size	int	End node size.
end_color	str	End color.
if_size	int	If node size.
if_color	str	If color.
oneof_size	int	OneOf node size.
oneof_color	str	OneOf color.
pipeline_size	int	Pipeline node size.
pipeline_color	str	Pipeline color.
caption_size	Literal[1, 2, 3]	Caption size.

Source code in python/sdag/visualization.py

class DAGViewer:
    """DAG viewer.

    Attributes:
        squeeze (bool): Collapse all inner pipelines into
            a unique node.
        task_size (int, optional): Task node size.
        task_color (str, optional): Task color.
        root_size (int, optional): Root node size.
        root_color (str, optional): Root color.
        end_size (int, optional): End node size.
        end_color (str, optional): End color.
        if_size (int, optional): If node size.
        if_color (str, optional): If color.
        oneof_size (int, optional): OneOf node size.
        oneof_color (str, optional): OneOf color.
        pipeline_size (int, optional): Pipeline node size.
        pipeline_color (str, optional): Pipeline color.
        caption_size (Literal[1, 2, 3], optional): Caption size.
    """

    def __init__(
        self,
        squeeze: bool = False,  # noqa: FBT002
        task_size: int = 50,
        task_color: str = "#3E7B7C",
        root_size: int = 15,
        root_color: str = "#F5F7FA",
        end_size: int = 15,
        end_color: str = "#F5F7FA",
        if_size: int = 30,
        if_color: str = "#D4A5A5",
        oneof_size: int = 30,
        oneof_color: str = "#D4A5A5",
        pipeline_size: int = 50,
        pipeline_color: str = "#F0D48D",
        caption_size: Literal[1, 2, 3] = 2,
    ) -> None:
        """Initialize the viewer.

        Args:
            squeeze (bool): Collapse subpipelines into a unique node.
                Defaults to False.
            task_size (int, optional): Task node size. Defaults to 50.
            task_color (str, optional): Task color.
                Defaults to "#3E7B7C".
            root_size (int, optional): Root node size. Defaults to 15.
            root_color (str, optional): Root color. Defaults to "#F5F7FA".
            end_size (int, optional): End node size. Defaults to 15.
            end_color (str, optional): End color. Defaults to "#F5F7FA".
            if_size (int, optional): If node size. Defaults to 30.
            if_color (str, optional): If color. Defaults to "#D4A5A5".
            oneof_size (int, optional): OneOf node size. Defaults to 30.
            oneof_color (str, optional): OneOf color. Defaults to "#D4A5A5".
            pipeline_size (int, optional): Pipeline size (only usable if
                squeeze is enabled). Defaults to 50.
            pipeline_color (str, optional): Pipeline color (only usable if
                squeeze is enabled). Defaults to "#F0D48D".
            caption_size (Literal[1, 2, 3], optional): Caption size.
                Defaults to 2.
        """
        self.squeeze = squeeze
        self.task_size = task_size
        self.task_color = task_color
        self.root_size = root_size
        self.root_color = root_color
        self.end_size = end_size
        self.end_color = end_color
        self.if_size = if_size
        self.if_color = if_color
        self.oneof_size = oneof_size
        self.oneof_color = oneof_color
        self.pipeline_size = pipeline_size
        self.pipeline_color = pipeline_color
        self.caption_size = caption_size

    def view(
        self, path: str, input_kwargs: dict[str, Any] | None = None
    ) -> "HTML":
        """View the DAG graph in a notebook cell.

        Args:
            path (str): Compiled pipeline path.
            input_kwargs (dict[str, Any] | None): Pipeline input kwargs,
                only used if the pipeline is compiled. Defaults to None.

        Returns:
            HTML: IPython HTML graph.
        """
        from neo4j_viz import Layout, VisualizationGraph
        from neo4j_viz import Node as VizNode

        from sdag.commands import _get_dag_from_name_import_or_json

        viz_nodes: list[VizNode] = []
        edges: list[Relationship] = []

        kwargs = input_kwargs if input_kwargs is not None else {}
        dag = _get_dag_from_name_import_or_json(
            path=path, input_kwargs=kwargs, extra_metadata=None
        )
        nodes = {node.uid: node for node in dag.nodes}
        for node in dag.nodes:
            if self._is_terminal(node):
                continue

            uid = self._get_uid(node)
            viz_node = self._get_viznode(node, uid)
            viz_nodes.append(viz_node)

            for parent in node.parents:
                parent_node = nodes[parent.uid]
                if self._is_terminal(parent_node):
                    continue

                if self.squeeze and parent_node.root_uid == node.root_uid != 0:
                    continue

                parent_uid = self._get_uid(parent_node)
                edge = self._get_edge(
                    source_uid=parent_uid, target_uid=uid, parent=parent
                )
                edges.append(edge)

        vg = VisualizationGraph(nodes=viz_nodes, relationships=edges)

        return vg.render(layout=Layout.HIERARCHICAL)

    def _is_terminal(self, node: NodeUnion) -> bool:
        """Check whether the node is terminal.

        Args:
            node (NodeUnion): Node.

        Returns:
            bool: True if the node is terminal.
        """
        return node.kind == "end" and node.root_uid == 0

    def _get_uid(self, node: NodeUnion) -> int:
        """Get the node uid (which might be the root one).

        Args:
            node (NodeUnion): Node.

        Returns:
            int: uid.
        """
        if self.squeeze and node.root_uid != 0:
            uid = node.root_uid
        else:
            uid = node.uid
        return uid

    def _get_viznode(self, node: NodeUnion, uid: int) -> "VizNode":
        """Get a visualization node.

        Args:
            node (NodeUnion): Node.
            uid (int): Node uid.

        Raises:
            ValueError: Unknown node kind.

        Returns:
            VizNode: Visualization node.
        """
        from neo4j_viz import Node as VizNode

        if self.squeeze and node.root_uid != 0:
            color = self.pipeline_color
            size = self.pipeline_size
            caption = node.pipeline_name
        else:
            match node.kind:
                case "task":
                    color = self.task_color
                    size = self.task_size
                    caption = node.name

                case "branch":
                    color = self.if_color
                    size = self.if_size
                    caption = "T/F"

                case "one_of":
                    color = self.oneof_color
                    size = self.oneof_size
                    caption = "one_of"

                case "root":
                    color = self.root_color
                    size = self.root_size
                    caption = ""

                case "end":
                    color = self.end_color
                    size = self.end_size
                    caption = ""

                case _:
                    msg = f"Unknown node type {node.type}"
                    raise ValueError(msg)

        return VizNode(
            id=uid,
            caption=caption,
            size=size,
            color=color,
            caption_size=self.caption_size,
        )  # type: ignore

    def _get_edge(
        self, source_uid: int, target_uid: int, parent: Parent
    ) -> "Relationship":
        """Get the relationship based on the relation type.

        Args:
            source_uid (int): Source node uid.
            target_uid (int): Target node uid.
            parent (Parent): Parent.

        Raises:
            ValueError: Unknown parent type.

        Returns:
            Relationship: Graph edge.
        """
        from neo4j_viz import Relationship

        match parent.kind.kind:
            case "logical":
                caption = ""
            case "output":
                caption = "output"
            case "branch":
                caption = "T" if parent.kind.branch else "F"  # type: ignore
            case "artifact":
                caption = parent.kind.name  # type: ignore
            case _:
                msg = f"Unknown parent kind {parent.kind.kind}"
                raise ValueError(msg)

        return Relationship(
            source=source_uid,  # type: ignore
            target=target_uid,
            caption=caption,
            caption_size=self.caption_size,
        )

__init__(squeeze=False, task_size=50, task_color='#3E7B7C', root_size=15, root_color='#F5F7FA', end_size=15, end_color='#F5F7FA', if_size=30, if_color='#D4A5A5', oneof_size=30, oneof_color='#D4A5A5', pipeline_size=50, pipeline_color='#F0D48D', caption_size=2)

Initialize the viewer.

Parameters:

Name	Type	Description	Default
squeeze	bool	Collapse subpipelines into a unique node. Defaults to False.	False
task_size	int	Task node size. Defaults to 50.	50
task_color	str	Task color. Defaults to "#3E7B7C".	'#3E7B7C'
root_size	int	Root node size. Defaults to 15.	15
root_color	str	Root color. Defaults to "#F5F7FA".	'#F5F7FA'
end_size	int	End node size. Defaults to 15.	15
end_color	str	End color. Defaults to "#F5F7FA".	'#F5F7FA'
if_size	int	If node size. Defaults to 30.	30
if_color	str	If color. Defaults to "#D4A5A5".	'#D4A5A5'
oneof_size	int	OneOf node size. Defaults to 30.	30
oneof_color	str	OneOf color. Defaults to "#D4A5A5".	'#D4A5A5'
pipeline_size	int	Pipeline size (only usable if squeeze is enabled). Defaults to 50.	50
pipeline_color	str	Pipeline color (only usable if squeeze is enabled). Defaults to "#F0D48D".	'#F0D48D'
caption_size	Literal[1, 2, 3]	Caption size. Defaults to 2.	2

Source code in python/sdag/visualization.py

def __init__(
    self,
    squeeze: bool = False,  # noqa: FBT002
    task_size: int = 50,
    task_color: str = "#3E7B7C",
    root_size: int = 15,
    root_color: str = "#F5F7FA",
    end_size: int = 15,
    end_color: str = "#F5F7FA",
    if_size: int = 30,
    if_color: str = "#D4A5A5",
    oneof_size: int = 30,
    oneof_color: str = "#D4A5A5",
    pipeline_size: int = 50,
    pipeline_color: str = "#F0D48D",
    caption_size: Literal[1, 2, 3] = 2,
) -> None:
    """Initialize the viewer.

    Args:
        squeeze (bool): Collapse subpipelines into a unique node.
            Defaults to False.
        task_size (int, optional): Task node size. Defaults to 50.
        task_color (str, optional): Task color.
            Defaults to "#3E7B7C".
        root_size (int, optional): Root node size. Defaults to 15.
        root_color (str, optional): Root color. Defaults to "#F5F7FA".
        end_size (int, optional): End node size. Defaults to 15.
        end_color (str, optional): End color. Defaults to "#F5F7FA".
        if_size (int, optional): If node size. Defaults to 30.
        if_color (str, optional): If color. Defaults to "#D4A5A5".
        oneof_size (int, optional): OneOf node size. Defaults to 30.
        oneof_color (str, optional): OneOf color. Defaults to "#D4A5A5".
        pipeline_size (int, optional): Pipeline size (only usable if
            squeeze is enabled). Defaults to 50.
        pipeline_color (str, optional): Pipeline color (only usable if
            squeeze is enabled). Defaults to "#F0D48D".
        caption_size (Literal[1, 2, 3], optional): Caption size.
            Defaults to 2.
    """
    self.squeeze = squeeze
    self.task_size = task_size
    self.task_color = task_color
    self.root_size = root_size
    self.root_color = root_color
    self.end_size = end_size
    self.end_color = end_color
    self.if_size = if_size
    self.if_color = if_color
    self.oneof_size = oneof_size
    self.oneof_color = oneof_color
    self.pipeline_size = pipeline_size
    self.pipeline_color = pipeline_color
    self.caption_size = caption_size

view(path, input_kwargs=None)

View the DAG graph in a notebook cell.

Parameters:

Name	Type	Description	Default
path	str	Compiled pipeline path.	required
input_kwargs	dict[str, Any] | None	Pipeline input kwargs, only used if the pipeline is compiled. Defaults to None.	None

Returns:

Name	Type	Description
HTML	HTML	IPython HTML graph.

Source code in python/sdag/visualization.py

def view(
    self, path: str, input_kwargs: dict[str, Any] | None = None
) -> "HTML":
    """View the DAG graph in a notebook cell.

    Args:
        path (str): Compiled pipeline path.
        input_kwargs (dict[str, Any] | None): Pipeline input kwargs,
            only used if the pipeline is compiled. Defaults to None.

    Returns:
        HTML: IPython HTML graph.
    """
    from neo4j_viz import Layout, VisualizationGraph
    from neo4j_viz import Node as VizNode

    from sdag.commands import _get_dag_from_name_import_or_json

    viz_nodes: list[VizNode] = []
    edges: list[Relationship] = []

    kwargs = input_kwargs if input_kwargs is not None else {}
    dag = _get_dag_from_name_import_or_json(
        path=path, input_kwargs=kwargs, extra_metadata=None
    )
    nodes = {node.uid: node for node in dag.nodes}
    for node in dag.nodes:
        if self._is_terminal(node):
            continue

        uid = self._get_uid(node)
        viz_node = self._get_viznode(node, uid)
        viz_nodes.append(viz_node)

        for parent in node.parents:
            parent_node = nodes[parent.uid]
            if self._is_terminal(parent_node):
                continue

            if self.squeeze and parent_node.root_uid == node.root_uid != 0:
                continue

            parent_uid = self._get_uid(parent_node)
            edge = self._get_edge(
                source_uid=parent_uid, target_uid=uid, parent=parent
            )
            edges.append(edge)

    vg = VisualizationGraph(nodes=viz_nodes, relationships=edges)

    return vg.render(layout=Layout.HIERARCHICAL)

MermaidGenerator

Generate a mermaid string of the DAG.

Attributes:

Name	Type	Description
squeeze	bool	Collapse all inner pipelines into a unique node.

Source code in python/sdag/visualization.py

class MermaidGenerator:
    """Generate a mermaid string of the DAG.

    Attributes:
        squeeze (bool): Collapse all inner pipelines into
            a unique node.
    """

    def __init__(self, squeeze: bool):
        """Initialize the mermaid generator.

        Args:
            squeeze (bool): Collapse all inner pipelines into
                a unique node.
        """
        self.squeeze = squeeze

    def generate_mermaid_string(self, dag: DAG) -> str:
        """Generate the mermaid string out of the dag.

        Args:
            dag (DAG): DAG.

        Returns:
            str: Mermaid string.
        """
        nodes = {node.uid: node for node in dag.nodes}
        lines = ["flowchart TB"]
        for node in dag.nodes:
            if self._is_terminal(node):
                continue

            for parent in node.parents:
                parent_node = nodes[parent.uid]
                if self._is_terminal(parent_node):
                    continue

                if self.squeeze and parent_node.root_uid == node.root_uid != 0:
                    continue

                parent_name = self._get_name(parent_node)
                child_name = self._get_name(node)
                edge = self._get_edge(parent)
                line = f"{parent_name} -->{edge} {child_name}"
                lines.append(line)

        return "\n".join(lines)

    def _is_terminal(self, node: NodeUnion) -> bool:
        """Check if the node is terminal.

        Args:
            node (NodeUnion): Node.

        Returns:
            bool: True if the node is terminal.
        """
        return node.kind == "end" and node.root_uid == 0

    def _get_name(self, node: NodeUnion) -> str:
        """Get the node name in the chart.

        Args:
            node (NodeUnion): Node.

        Returns:
            str: Node name.
        """
        if self.squeeze and node.root_uid != 0:
            return (
                f"{node.root_uid}_{node.pipeline_name}[{node.pipeline_name}]"
            )

        match node.kind:
            case "task":
                return f"{node.uid}[{node.name}]"
            case "branch":
                return f"{node.uid}{{T/F}}"
            case "root":
                return f"{node.uid}(_root_)"
            case "end":
                return f"{node.uid}(_end_)"
            case "one_of":
                return f"{node.uid}[oneof]"

    def _get_edge(self, parent: Parent) -> str:
        """Get the edge.

        Args:
            parent (Parent): Parent.

        Returns:
            str: Edge string.
        """
        match parent.kind.kind:
            case "logical":
                return ""
            case "output":
                return "|output|"
            case "branch":
                branch = "T" if parent.kind.branch else "F"  # type: ignore
                return f"|{branch}|"
            case "artifact":
                return f"|{parent.kind.name}|"  # type: ignore

__init__(squeeze)

Initialize the mermaid generator.

Parameters:

Name	Type	Description	Default
squeeze	bool	Collapse all inner pipelines into a unique node.	required

Source code in python/sdag/visualization.py

def __init__(self, squeeze: bool):
    """Initialize the mermaid generator.

    Args:
        squeeze (bool): Collapse all inner pipelines into
            a unique node.
    """
    self.squeeze = squeeze

generate_mermaid_string(dag)

Generate the mermaid string out of the dag.

Parameters:

Name	Type	Description	Default
dag	DAG	DAG.	required

Returns:

Name	Type	Description
str	str	Mermaid string.

Source code in python/sdag/visualization.py

def generate_mermaid_string(self, dag: DAG) -> str:
    """Generate the mermaid string out of the dag.

    Args:
        dag (DAG): DAG.

    Returns:
        str: Mermaid string.
    """
    nodes = {node.uid: node for node in dag.nodes}
    lines = ["flowchart TB"]
    for node in dag.nodes:
        if self._is_terminal(node):
            continue

        for parent in node.parents:
            parent_node = nodes[parent.uid]
            if self._is_terminal(parent_node):
                continue

            if self.squeeze and parent_node.root_uid == node.root_uid != 0:
                continue

            parent_name = self._get_name(parent_node)
            child_name = self._get_name(node)
            edge = self._get_edge(parent)
            line = f"{parent_name} -->{edge} {child_name}"
            lines.append(line)

    return "\n".join(lines)

wrappers

Wrappers.

Elif

Elif Wrapper.

Must follow an upstream If or Elif clause in the same scope, with nothing but plain task calls in between.

Attributes:

Name	Type	Description
_expr	TaskNode | OneOfNode	Task returning a bool output.
branch	BranchNode	Branch.

Source code in python/sdag/wrappers.py

class Elif:
    """Elif Wrapper.

    Must follow an upstream If or Elif clause in the same scope, with
    nothing but plain task calls in between.

    Attributes:
        _expr (TaskNode | OneOfNode): Task returning a bool output.
        branch (BranchNode): Branch.
    """

    def __init__(self, expr: TaskNode | OneOfNode):
        """Initialize the Elif wrapper.

        Args:
            expr (TaskNode | OneOfNode): Task returning a bool output.

        Raises:
            IncorrectElifError: Not immediately preceded by an If or
                Elif clause in the same scope.
        """
        prev = compiler.current_scope.last_clause
        if prev is None or prev.terminal:
            raise IncorrectElifError

        expr.add_branch_edge(prev.branch_uid, prev.next_edge_value)

        parent = Parent(uid=expr.uid, kind=OutputParent(key="expr"))
        self.branch = BranchNode(uid=compiler.get_uid(), parents=[parent])

    def __enter__(self) -> Self:
        """Enter the positive branch.

        Returns:
            Self: This task.
        """
        compiler.register(self.branch)
        compiler.push_scope((self.branch.uid, True))
        return self

    def __exit__(self, type, value, traceback) -> None:  # noqa: A002
        """Exit the positive branch."""
        compiler.pop_scope()
        compiler.close_clause(self.branch.uid, next_edge_value=False)

__enter__()

Enter the positive branch.

Returns:

Name	Type	Description
Self	Self	This task.

Source code in python/sdag/wrappers.py

def __enter__(self) -> Self:
    """Enter the positive branch.

    Returns:
        Self: This task.
    """
    compiler.register(self.branch)
    compiler.push_scope((self.branch.uid, True))
    return self

__exit__(type, value, traceback)

Exit the positive branch.

Source code in python/sdag/wrappers.py

def __exit__(self, type, value, traceback) -> None:  # noqa: A002
    """Exit the positive branch."""
    compiler.pop_scope()
    compiler.close_clause(self.branch.uid, next_edge_value=False)

__init__(expr)

Initialize the Elif wrapper.

Parameters:

Name	Type	Description	Default
expr	TaskNode | OneOfNode	Task returning a bool output.	required

Raises:

Type	Description
IncorrectElifError	Not immediately preceded by an If or Elif clause in the same scope.

Source code in python/sdag/wrappers.py

def __init__(self, expr: TaskNode | OneOfNode):
    """Initialize the Elif wrapper.

    Args:
        expr (TaskNode | OneOfNode): Task returning a bool output.

    Raises:
        IncorrectElifError: Not immediately preceded by an If or
            Elif clause in the same scope.
    """
    prev = compiler.current_scope.last_clause
    if prev is None or prev.terminal:
        raise IncorrectElifError

    expr.add_branch_edge(prev.branch_uid, prev.next_edge_value)

    parent = Parent(uid=expr.uid, kind=OutputParent(key="expr"))
    self.branch = BranchNode(uid=compiler.get_uid(), parents=[parent])

Else

Else wrapper.

Must follow an upstream If or Elif clause in the same scope, with nothing but plain task calls in between.

Source code in python/sdag/wrappers.py

class Else:
    """Else wrapper.

    Must follow an upstream If or Elif clause in the same scope, with
    nothing but plain task calls in between.
    """

    def __enter__(self) -> Self:
        """Enter the else context.

        Raises:
            IncorrectElseError: Not immediately preceded by an If or
                Elif clause in the same scope.

        Returns:
            Self: Else.
        """
        prev = compiler.current_scope.last_clause
        if prev is None or prev.terminal:
            raise IncorrectElseError

        self._branch_uid = prev.branch_uid
        compiler.push_scope((prev.branch_uid, prev.next_edge_value))
        return self

    def __exit__(self, exc_type, exc, tb):
        """Exit the context."""
        compiler.pop_scope()
        compiler.close_clause(
            self._branch_uid, next_edge_value=False, terminal=True
        )

__enter__()

Enter the else context.

Raises:

Type	Description
IncorrectElseError	Not immediately preceded by an If or Elif clause in the same scope.

Returns:

Name	Type	Description
Self	Self	Else.

Source code in python/sdag/wrappers.py

def __enter__(self) -> Self:
    """Enter the else context.

    Raises:
        IncorrectElseError: Not immediately preceded by an If or
            Elif clause in the same scope.

    Returns:
        Self: Else.
    """
    prev = compiler.current_scope.last_clause
    if prev is None or prev.terminal:
        raise IncorrectElseError

    self._branch_uid = prev.branch_uid
    compiler.push_scope((prev.branch_uid, prev.next_edge_value))
    return self

__exit__(exc_type, exc, tb)

Exit the context.

Source code in python/sdag/wrappers.py

def __exit__(self, exc_type, exc, tb):
    """Exit the context."""
    compiler.pop_scope()
    compiler.close_clause(
        self._branch_uid, next_edge_value=False, terminal=True
    )

If

Branch wrapper.

Attributes:

Name	Type	Description
branch	BranchNode	Branch.

Source code in python/sdag/wrappers.py

class If:
    """Branch wrapper.

    Attributes:
        branch (BranchNode): Branch.
    """

    def __init__(self, expr: TaskNode | OneOfNode):
        """Initialize the branch wrapper.

        Args:
            expr (TaskNode | OneOfNode): Task returning a bool output.
        """
        parent = Parent(uid=expr.uid, kind=OutputParent(key="expr"))
        self.branch = BranchNode(uid=compiler.get_uid(), parents=[parent])

    def __enter__(self) -> Self:
        """Enter the positive branch.

        Returns:
            Self: This task.
        """
        compiler.register(self.branch)
        compiler.push_scope((self.branch.uid, True))
        return self

    def __exit__(self, type, value, traceback) -> None:  # noqa: A002
        """Exit the positive branch."""
        compiler.pop_scope()
        compiler.close_clause(self.branch.uid, next_edge_value=False)

__enter__()

Enter the positive branch.

Returns:

Name	Type	Description
Self	Self	This task.

Source code in python/sdag/wrappers.py

def __enter__(self) -> Self:
    """Enter the positive branch.

    Returns:
        Self: This task.
    """
    compiler.register(self.branch)
    compiler.push_scope((self.branch.uid, True))
    return self

__exit__(type, value, traceback)

Exit the positive branch.

Source code in python/sdag/wrappers.py

def __exit__(self, type, value, traceback) -> None:  # noqa: A002
    """Exit the positive branch."""
    compiler.pop_scope()
    compiler.close_clause(self.branch.uid, next_edge_value=False)

__init__(expr)

Initialize the branch wrapper.

Parameters:

Name	Type	Description	Default
expr	TaskNode | OneOfNode	Task returning a bool output.	required

Source code in python/sdag/wrappers.py

def __init__(self, expr: TaskNode | OneOfNode):
    """Initialize the branch wrapper.

    Args:
        expr (TaskNode | OneOfNode): Task returning a bool output.
    """
    parent = Parent(uid=expr.uid, kind=OutputParent(key="expr"))
    self.branch = BranchNode(uid=compiler.get_uid(), parents=[parent])

Pipeline

Pipeline.

Attributes:

Name	Type	Description
fn	Callable[..., Any]	Pipeline function.
tasks	dict[str, Task]	dict[str, Task]: Pipeline local tasks.

Source code in python/sdag/wrappers.py

class Pipeline:
    """Pipeline.

    Attributes:
        fn (Callable[..., Any]): Pipeline function.
        tasks: dict[str, Task]: Pipeline local tasks.
    """

    def __init__(self, fn: Callable[..., Any]):
        """Initialize the pipeline wrapper.

        Args:
            fn (Callable[..., Any]): Pipeline function.
        """
        self.fn = fn
        self.tasks: dict[str, Task] = {}

    def add_task(self, task: Task) -> None:
        """Add a local task.

        Args:
            task (Task): Task local to the pipeline.

        Raises:
            TaskNotUniqueError: A task with the same name
                already exists.
        """
        if task.fn.__name__ in self.tasks:
            raise TaskNotUniqueError(name=task.fn.__name__)
        self.tasks[task.fn.__name__] = task

    def task(
        self,
        script: str | Path | Script,
        name: str | None = None,
        cmd: Commands = "sbatch",
        mode: ExecMode = "wrap",
        cache: CacheOptions = "none",
        cache_ignore: list[str] | None = None,
        cache_size: int = 1,
        njobs: int = 1,
        retries: int = 0,
        tags: list[str] | None = None,
    ) -> Callable[[Callable], Task]:
        """Local task decorator.

        Args:
            script (str | Path | Script): Launch script. It can be
                either or a sdag.Script instance.
            name (str | None, optional): Task name. If null, it will
                default to the function name. Defaults to None.
            cmd (Commands, optional): Command to
                execute the script. Defaults to "sbatch".
            mode (ExecMode, optional): Use wrap to call the
                Python function or ext to call an external script.
                Defaults to "wrap".
            cache (CacheOptions, optional): Local caching options.
                'none': No caching. 'output': All output artifacts
                must exist. 'io': Input values must match and all
                output artifacts exist. 'task': Same as 'io'
                but the task function code is also checked. 'project':
                Same as 'io' but the code of all internal modules
                imported when the pipeline is compiled must also match.
                Defaults to 'none'.
            cache_ignore (list[str] | None): List of input arguments
                ignored during cache validation. Ignored if cache is
                set to 'none' or 'output'. Defaults to None.
            cache_size (int): Cache size. Ignore if caching is
                disabled. set to 0 to allow for infinite cache
                size.
            njobs (int): When spreading a collection over multiple
                jobs with each(), njobs represents the number of
                jobs in the array. It is ignored if the task is not
                an array. Defaults to 1.
            retries: (int): Number of retries. Defaults to 0.
            tags (list[str] | None): Task tags, they can be used to
                configure sets of tasks globally.

        Returns:
            Callable[[Callable], Task]: Task wrapper.
        """

        def return_task(fn: Callable[..., Any]) -> Task:
            """Get the task.

            Args:
                fn (Callable[..., Any]): Task function.

            Returns:
                Task: Task.
            """
            task_name = name if name is not None else fn.__name__
            if isinstance(script, Script):
                script_obj = ScriptContent(content=script.content)
            else:
                script_obj = ScriptPath(path=Path(script))

            task = Task(
                fn=fn,
                name=task_name,
                scope="local",
                cmd=cmd,
                mode=mode,
                cache=cache,
                cache_ignore=cache_ignore,
                cache_size=cache_size,
                njobs=njobs,
                retries=retries,
                script=script_obj,
                tags=tags if tags is not None else [],
            )
            self.add_task(task)

            return task

        return return_task

    def __call__(self, *args: NodeUnion, **input_kwargs: Any) -> EndNode:
        """Call the pipeline function to compile it.

        Args:
            args (Node): Parent nodes.
            input_kwargs (Any): Pipeline input kwargs.

        Returns:
            Node: End node of the pipeline.
        """
        root = RootNode(uid=compiler.get_uid(), pipeline_name=self.fn.__name__)
        for parent in args:
            root.add_logical_edge(parent.uid)

        compiler.set_dag(root)
        self.fn(**input_kwargs)
        dag, root, end = compiler.get_dag()

        for node in dag.nodes:
            if node.kind in ("one_of", "task"):
                end.join_artifacts(node)  # type: ignore

        return end

    def compile(self, import_path: str | None = None, **kwargs: Any) -> DAG:
        """Call the pipeline function to compile it.

        Args:
            import_path (str | None): Pipeline import path. If null, it will
                be set to the pipeline function name. Defaults to None.
            kwargs (Any): Pipeline input kwargs.

        Returns:
            Node: End node of the pipeline.
        """
        compiler.reset()
        root = RootNode(uid=compiler.get_uid(), pipeline_name=self.fn.__name__)
        compiler.set_dag(root, import_path)
        self.fn(**kwargs)
        dag, _, _ = compiler.get_dag()

        return dag

    def compute_code_hash(self) -> str:
        """Compute the hash of all internal imported modules.

        Returns:
            str: Computed hash, it's an empty string if no
                modules are found.
        """
        hasher = ModuleHasher()
        return hasher.hash_modules()

__call__(*args, **input_kwargs)

Call the pipeline function to compile it.

Parameters:

Name	Type	Description	Default
args	Node	Parent nodes.	()
input_kwargs	Any	Pipeline input kwargs.	{}

Returns:

Name	Type	Description
Node	EndNode	End node of the pipeline.

Source code in python/sdag/wrappers.py

def __call__(self, *args: NodeUnion, **input_kwargs: Any) -> EndNode:
    """Call the pipeline function to compile it.

    Args:
        args (Node): Parent nodes.
        input_kwargs (Any): Pipeline input kwargs.

    Returns:
        Node: End node of the pipeline.
    """
    root = RootNode(uid=compiler.get_uid(), pipeline_name=self.fn.__name__)
    for parent in args:
        root.add_logical_edge(parent.uid)

    compiler.set_dag(root)
    self.fn(**input_kwargs)
    dag, root, end = compiler.get_dag()

    for node in dag.nodes:
        if node.kind in ("one_of", "task"):
            end.join_artifacts(node)  # type: ignore

    return end

__init__(fn)

Initialize the pipeline wrapper.

Parameters:

Name	Type	Description	Default
fn	Callable[..., Any]	Pipeline function.	required

Source code in python/sdag/wrappers.py

def __init__(self, fn: Callable[..., Any]):
    """Initialize the pipeline wrapper.

    Args:
        fn (Callable[..., Any]): Pipeline function.
    """
    self.fn = fn
    self.tasks: dict[str, Task] = {}

add_task(task)

Add a local task.

Parameters:

Name	Type	Description	Default
task	Task	Task local to the pipeline.	required

Raises:

Type	Description
TaskNotUniqueError	A task with the same name already exists.

Source code in python/sdag/wrappers.py

def add_task(self, task: Task) -> None:
    """Add a local task.

    Args:
        task (Task): Task local to the pipeline.

    Raises:
        TaskNotUniqueError: A task with the same name
            already exists.
    """
    if task.fn.__name__ in self.tasks:
        raise TaskNotUniqueError(name=task.fn.__name__)
    self.tasks[task.fn.__name__] = task

compile(import_path=None, **kwargs)

Call the pipeline function to compile it.

Parameters:

Name	Type	Description	Default
import_path	str | None	Pipeline import path. If null, it will be set to the pipeline function name. Defaults to None.	None
kwargs	Any	Pipeline input kwargs.	{}

Returns:

Name	Type	Description
Node	DAG	End node of the pipeline.

Source code in python/sdag/wrappers.py

def compile(self, import_path: str | None = None, **kwargs: Any) -> DAG:
    """Call the pipeline function to compile it.

    Args:
        import_path (str | None): Pipeline import path. If null, it will
            be set to the pipeline function name. Defaults to None.
        kwargs (Any): Pipeline input kwargs.

    Returns:
        Node: End node of the pipeline.
    """
    compiler.reset()
    root = RootNode(uid=compiler.get_uid(), pipeline_name=self.fn.__name__)
    compiler.set_dag(root, import_path)
    self.fn(**kwargs)
    dag, _, _ = compiler.get_dag()

    return dag

compute_code_hash()

Compute the hash of all internal imported modules.

Returns:

Name	Type	Description
str	str	Computed hash, it's an empty string if no modules are found.

Source code in python/sdag/wrappers.py

def compute_code_hash(self) -> str:
    """Compute the hash of all internal imported modules.

    Returns:
        str: Computed hash, it's an empty string if no
            modules are found.
    """
    hasher = ModuleHasher()
    return hasher.hash_modules()

task(script, name=None, cmd='sbatch', mode='wrap', cache='none', cache_ignore=None, cache_size=1, njobs=1, retries=0, tags=None)

Local task decorator.

Parameters:

Name	Type	Description	Default
script	str | Path | Script	Launch script. It can be either or a sdag.Script instance.	required
name	str | None	Task name. If null, it will default to the function name. Defaults to None.	None
cmd	Commands	Command to execute the script. Defaults to "sbatch".	'sbatch'
mode	ExecMode	Use wrap to call the Python function or ext to call an external script. Defaults to "wrap".	'wrap'
cache	CacheOptions	Local caching options. 'none': No caching. 'output': All output artifacts must exist. 'io': Input values must match and all output artifacts exist. 'task': Same as 'io' but the task function code is also checked. 'project': Same as 'io' but the code of all internal modules imported when the pipeline is compiled must also match. Defaults to 'none'.	'none'
cache_ignore	list[str] | None	List of input arguments ignored during cache validation. Ignored if cache is set to 'none' or 'output'. Defaults to None.	None
cache_size	int	Cache size. Ignore if caching is disabled. set to 0 to allow for infinite cache size.	1
njobs	int	When spreading a collection over multiple jobs with each(), njobs represents the number of jobs in the array. It is ignored if the task is not an array. Defaults to 1.	1
retries	int	(int): Number of retries. Defaults to 0.	0
tags	list[str] | None	Task tags, they can be used to configure sets of tasks globally.	None

Returns:

Type	Description
Callable[[Callable], Task]	Callable[[Callable], Task]: Task wrapper.

Source code in python/sdag/wrappers.py

def task(
    self,
    script: str | Path | Script,
    name: str | None = None,
    cmd: Commands = "sbatch",
    mode: ExecMode = "wrap",
    cache: CacheOptions = "none",
    cache_ignore: list[str] | None = None,
    cache_size: int = 1,
    njobs: int = 1,
    retries: int = 0,
    tags: list[str] | None = None,
) -> Callable[[Callable], Task]:
    """Local task decorator.

    Args:
        script (str | Path | Script): Launch script. It can be
            either or a sdag.Script instance.
        name (str | None, optional): Task name. If null, it will
            default to the function name. Defaults to None.
        cmd (Commands, optional): Command to
            execute the script. Defaults to "sbatch".
        mode (ExecMode, optional): Use wrap to call the
            Python function or ext to call an external script.
            Defaults to "wrap".
        cache (CacheOptions, optional): Local caching options.
            'none': No caching. 'output': All output artifacts
            must exist. 'io': Input values must match and all
            output artifacts exist. 'task': Same as 'io'
            but the task function code is also checked. 'project':
            Same as 'io' but the code of all internal modules
            imported when the pipeline is compiled must also match.
            Defaults to 'none'.
        cache_ignore (list[str] | None): List of input arguments
            ignored during cache validation. Ignored if cache is
            set to 'none' or 'output'. Defaults to None.
        cache_size (int): Cache size. Ignore if caching is
            disabled. set to 0 to allow for infinite cache
            size.
        njobs (int): When spreading a collection over multiple
            jobs with each(), njobs represents the number of
            jobs in the array. It is ignored if the task is not
            an array. Defaults to 1.
        retries: (int): Number of retries. Defaults to 0.
        tags (list[str] | None): Task tags, they can be used to
            configure sets of tasks globally.

    Returns:
        Callable[[Callable], Task]: Task wrapper.
    """

    def return_task(fn: Callable[..., Any]) -> Task:
        """Get the task.

        Args:
            fn (Callable[..., Any]): Task function.

        Returns:
            Task: Task.
        """
        task_name = name if name is not None else fn.__name__
        if isinstance(script, Script):
            script_obj = ScriptContent(content=script.content)
        else:
            script_obj = ScriptPath(path=Path(script))

        task = Task(
            fn=fn,
            name=task_name,
            scope="local",
            cmd=cmd,
            mode=mode,
            cache=cache,
            cache_ignore=cache_ignore,
            cache_size=cache_size,
            njobs=njobs,
            retries=retries,
            script=script_obj,
            tags=tags if tags is not None else [],
        )
        self.add_task(task)

        return task

    return return_task

Script

Script content.

It can be used to embed the script content in the pipeline definition.

Attributes:

Name	Type	Description
content	str	Script content.

Source code in python/sdag/wrappers.py

class Script:
    """Script content.

    It can be used to embed the script content
    in the pipeline definition.

    Attributes:
        content (str): Script content.
    """

    def __init__(self, content: str):
        """Initialize the Script.

        Args:
            content (str): Script content.
        """
        self.content = content

__init__(content)

Initialize the Script.

Parameters:

Name	Type	Description	Default
content	str	Script content.	required

Source code in python/sdag/wrappers.py

def __init__(self, content: str):
    """Initialize the Script.

    Args:
        content (str): Script content.
    """
    self.content = content

Task

Task wrapper.

Attributes:

Name	Type	Description
fn	Callable[..., Any]	Task function.
name	str	Task name.
cmd	Commands	Task command.
mode	ExecMode	Task mode.
scope	Scope	(Scope): Task scope. Use 'global' for tasks decorated with the @task decorator, 'local' for tasks decorated with @pipeline.task.
cache	bool	Enable caching.
cache_ignore	list[str]	list of fields ignored during cache validation.
cache_size	int	Cache size. Ignore if caching is disabled. set to 0 to allow for infinite cache size.
njobs	int	Number of jobs in the array.
retries	int	Number of retries.
script	ScriptUnion	Submission script.

Source code in python/sdag/wrappers.py

class Task:
    """Task wrapper.

    Attributes:
        fn (Callable[..., Any]): Task function.
        name (str): Task name.
        cmd (Commands): Task command.
        mode (ExecMode): Task mode.
        scope: (Scope): Task scope.
            Use 'global' for tasks decorated with the
            @task decorator, 'local' for tasks decorated
            with @pipeline.task.
        cache (bool): Enable caching.
        cache_ignore (list[str]): list of fields ignored
            during cache validation.
        cache_size (int): Cache size. Ignore if caching is disabled.
            set to 0 to allow for infinite cache size.
        njobs (int): Number of jobs in the array.
        retries (int): Number of retries.
        script (ScriptUnion): Submission script.
    """

    def __init__(
        self,
        fn: Callable[..., Any],
        name: str,
        cmd: Commands,
        mode: ExecMode,
        scope: Scope,
        cache: CacheOptions,
        cache_ignore: list[str] | None,
        cache_size: int,
        njobs: int,
        retries: int,
        script: ScriptUnion,
        tags: list[str],
    ):
        """Initialize the task wrapper.

        Args:
            fn (Callable[..., Any]): Task function.
            name (str): Task name.
            cmd (Commands): Task command.
            mode (ExecMode): Task mode.
            scope: (Scope): Task scope.
                Use 'global' for tasks decorated with the
                @task decorator, 'local' for tasks decorated
                with @pipeline.task.
            cache (CacheOptions): Caching options.
            cache_ignore (list[str] | None): list of fields
                ignored during cache validation.
            cache_size (int): Cache size. Ignore if caching is
                disabled. set to 0 to allow for infinite cache
                size.
            njobs (int): When spreading a collection over multiple
                jobs with each(), njobs represents the number of
                jobs in the array. Defaults to 1.
            retries (int): Number of retries.
            script (ScriptUnion): Submission script.
            tags (list[str]): Tags.
        """
        self.fn = fn
        self.name = name
        self.cmd: Commands = cmd
        self.mode: ExecMode = mode
        self.scope: Scope = scope
        self.cache: CacheOptions = cache
        self.cache_ignore = cache_ignore if cache_ignore is not None else []
        self.cache_size = cache_size
        self.njobs = njobs
        self.retries = retries
        self.script = script
        self.tags = tags

    def __call__(self, *args: NodeUnion, **kwargs: Any) -> TaskNode:
        """Call the task to get a node in the graph.

        Args:
            args (NodeUnion): Logical dependencies.
            kwargs (Any): Kwargs, outputs, and artifacts.

        Returns:
            TaskNode: Task node.
        """
        node = TaskNode(
            uid=compiler.get_uid(),
            fn_name=self.fn.__name__,
            name=self.name,
            scope=self.scope,
            cache=self.cache,
            cache_ignore=self.cache_ignore.copy(),
            cache_size=self.cache_size,
            code_hash=self.code_hash,
            mode=self.mode,
            cmd=self.cmd,
            njobs=self.njobs,
            retries=self.retries,
            script=self.script,
            tags=self.tags.copy(),
        )

        for parent in args:
            node.add_logical_edge(parent.uid)

        self._add_kwargs(node, kwargs)
        compiler.register(node)

        return node

    def _add_kwargs(self, node: TaskNode, kwargs: dict[str, Any]) -> None:
        """Add kwargs.

        Args:
            node (TaskNode): Task node.
            kwargs (dict[str, Any]): Input kwargs.

        Raises:
            KwargNotFoundError: Keyword argument not found in
                the input values.
        """
        sig = inspect.signature(self.fn)
        self._add_default_values(sig, kwargs)
        validate_kwargs(sig, kwargs)
        self._prepend_base_path_if_set(kwargs, sig)

        for key, val in kwargs.items():
            param = sig.parameters.get(key)
            if is_artifact(param):
                self._set_output_artifact(node, key, val)

            val = kwargs[key]
            self._handle_input_kwarg(node, key, val)

    def _prepend_base_path_if_set(
        self, kwargs: dict[str, Any], sig: inspect.Signature
    ) -> None:
        """Prepend the base path if required.

        If the `SDAG_BASE_PATH` environment variable is set and
        path is not absolute, the registered path is appended to
        the base path.

        Args:
            kwargs (dict[str, Any]): Input kwargs.
            sig (inspect.Signature): Task signature.
        """
        settings = get_compile_settings()
        if settings.sdag_base_path is None:
            return

        logger.debug("sdag base path: '%s'", settings.sdag_base_path)

        for key in kwargs:
            param = sig.parameters.get(key)
            if is_artifact(param):
                path = Path(kwargs[key])
                if not path.is_absolute():
                    kwargs[key] = str(settings.sdag_base_path / path)

    def _add_default_values(
        self, sig: inspect.Signature, kwargs: dict[str, Any]
    ) -> None:
        """Add default values to missing kwargs.

        Args:
            sig (inspect.Signature): Function signature.
            kwargs (dict[str, Any]): Input kwargs.
        """
        for key, param in sig.parameters.items():
            if key not in kwargs and param.default is not param.empty:
                kwargs[key] = param.default

    def _set_output_artifact(
        self, node: TaskNode, key: str, value: Any
    ) -> None:
        """Set output artifact.

        Args:
            node (TaskNode): Task node.
            key (str): Artifact key.
            value (Any): Input value.

        Raises:
            DynamicArtifactError: Dynamic artifacts are not supported.
        """
        if isinstance(value, NodeUnion.__args__[0]):  # type: ignore
            raise DynamicArtifactError(key)

        path = (
            value.path if isinstance(value, ArtifactContainer) else Path(value)
        )
        node.register_artifact(key, path)

    def _handle_input_kwarg(
        self, node: TaskNode, key: str, value: Any
    ) -> None:
        """Handle input kwargs.

        Three types of input kwargs are possible:
        - dynamic input
        - artifacts
        - static input

        Args:
            node (TaskNode): Child node.
            key (str): Input key.
            value (Any): Input value.
        """
        if isinstance(value, ArtifactContainer):
            node.add_artifact_edge(
                parent_uid=value.node.uid,
                key=key,
                name=value.key,
                path=value.path,
            )

        else:
            if isinstance(value, Each):
                if node.each is not None:
                    raise MultipleEachError(key, node.each)
                node.each = key
                value = value.collection

            if isinstance(value, BaseNode):
                node.add_output_edge(value.uid, key)

            else:
                input_kwarg = Kwarg(key=key, value=value)
                node.add_kwarg(input_kwarg)

    @cached_property
    def code_hash(self) -> str:
        """Compute and cache the source code hash.

        Returns:
            str: Hash of the task function source code.
        """
        import hashlib

        source = inspect.getsource(self.fn)
        return hashlib.sha256(source.encode("utf-8")).hexdigest()

code_hash cached property

Compute and cache the source code hash.

Returns:

Name	Type	Description
str	str	Hash of the task function source code.

__call__(*args, **kwargs)

Call the task to get a node in the graph.

Parameters:

Name	Type	Description	Default
args	NodeUnion	Logical dependencies.	()
kwargs	Any	Kwargs, outputs, and artifacts.	{}

Returns:

Name	Type	Description
TaskNode	TaskNode	Task node.

Source code in python/sdag/wrappers.py

def __call__(self, *args: NodeUnion, **kwargs: Any) -> TaskNode:
    """Call the task to get a node in the graph.

    Args:
        args (NodeUnion): Logical dependencies.
        kwargs (Any): Kwargs, outputs, and artifacts.

    Returns:
        TaskNode: Task node.
    """
    node = TaskNode(
        uid=compiler.get_uid(),
        fn_name=self.fn.__name__,
        name=self.name,
        scope=self.scope,
        cache=self.cache,
        cache_ignore=self.cache_ignore.copy(),
        cache_size=self.cache_size,
        code_hash=self.code_hash,
        mode=self.mode,
        cmd=self.cmd,
        njobs=self.njobs,
        retries=self.retries,
        script=self.script,
        tags=self.tags.copy(),
    )

    for parent in args:
        node.add_logical_edge(parent.uid)

    self._add_kwargs(node, kwargs)
    compiler.register(node)

    return node

__init__(fn, name, cmd, mode, scope, cache, cache_ignore, cache_size, njobs, retries, script, tags)

Initialize the task wrapper.

Parameters:

Name	Type	Description	Default
fn	Callable[..., Any]	Task function.	required
name	str	Task name.	required
cmd	Commands	Task command.	required
mode	ExecMode	Task mode.	required
scope	Scope	(Scope): Task scope. Use 'global' for tasks decorated with the @task decorator, 'local' for tasks decorated with @pipeline.task.	required
cache	CacheOptions	Caching options.	required
cache_ignore	list[str] | None	list of fields ignored during cache validation.	required
cache_size	int	Cache size. Ignore if caching is disabled. set to 0 to allow for infinite cache size.	required
njobs	int	When spreading a collection over multiple jobs with each(), njobs represents the number of jobs in the array. Defaults to 1.	required
retries	int	Number of retries.	required
script	ScriptUnion	Submission script.	required
tags	list[str]	Tags.	required

Source code in python/sdag/wrappers.py

def __init__(
    self,
    fn: Callable[..., Any],
    name: str,
    cmd: Commands,
    mode: ExecMode,
    scope: Scope,
    cache: CacheOptions,
    cache_ignore: list[str] | None,
    cache_size: int,
    njobs: int,
    retries: int,
    script: ScriptUnion,
    tags: list[str],
):
    """Initialize the task wrapper.

    Args:
        fn (Callable[..., Any]): Task function.
        name (str): Task name.
        cmd (Commands): Task command.
        mode (ExecMode): Task mode.
        scope: (Scope): Task scope.
            Use 'global' for tasks decorated with the
            @task decorator, 'local' for tasks decorated
            with @pipeline.task.
        cache (CacheOptions): Caching options.
        cache_ignore (list[str] | None): list of fields
            ignored during cache validation.
        cache_size (int): Cache size. Ignore if caching is
            disabled. set to 0 to allow for infinite cache
            size.
        njobs (int): When spreading a collection over multiple
            jobs with each(), njobs represents the number of
            jobs in the array. Defaults to 1.
        retries (int): Number of retries.
        script (ScriptUnion): Submission script.
        tags (list[str]): Tags.
    """
    self.fn = fn
    self.name = name
    self.cmd: Commands = cmd
    self.mode: ExecMode = mode
    self.scope: Scope = scope
    self.cache: CacheOptions = cache
    self.cache_ignore = cache_ignore if cache_ignore is not None else []
    self.cache_size = cache_size
    self.njobs = njobs
    self.retries = retries
    self.script = script
    self.tags = tags

each(collection)

Job array collection.

Used to spread a collection over an array of jobs. The input can be either a compile-time, static list or a list coming as the output of a task.

Parameters:

Name	Type	Description	Default
collection	list[Any] | TaskNode	Collection that will be spread across jobs.	required

Returns:

Name	Type	Description
Each	Each	Collection wrapper.

Source code in python/sdag/wrappers.py

def each(collection: list[Any] | TaskNode) -> Each:
    """Job array collection.

    Used to spread a collection over an array of jobs. The
    input can be either a compile-time, static list or
    a list coming as the output of a task.

    Args:
        collection (list[Any] | TaskNode): Collection that
            will be spread across jobs.

    Returns:
        Each: Collection wrapper.
    """
    return Each(collection)

is_artifact(param)

Check if a type hint is actually an artifact.

Parameters:

Name	Type	Description	Default
param	Parameter | None	Parameter.	required

Returns:

Name	Type	Description
bool	bool	True if it's an artifact.

Source code in python/sdag/wrappers.py

def is_artifact(param: inspect.Parameter | None) -> bool:
    """Check if a type hint is actually an artifact.

    Args:
        param (inspect.Parameter | None): Parameter.

    Returns:
        bool: True if it's an artifact.
    """
    if param is None:
        return False
    annotation = param.annotation
    if not hasattr(annotation, "__metadata__"):
        return False
    meta = annotation.__metadata__
    return meta and meta[0] == ArtifactSig

oneof(*args, name='one_of')

OneOf node.

It has two major use cases: 1. Selecting nodes from mutually excluded branches. 2. Selecting The first successfully completed parent.

Returns:

Type	Description
OneOfNode	Node[OneOfNode]: OneOf node.

Source code in python/sdag/wrappers.py

def oneof(*args: NodeUnion, name: str = "one_of") -> OneOfNode:
    """OneOf node.

    It has two major use cases:
    1. Selecting nodes from mutually excluded branches.
    2. Selecting The first successfully completed parent.

    Returns:
        Node[OneOfNode]: OneOf node.
    """
    node = OneOfNode(uid=compiler.get_uid(), name=name)
    compiler.register(node)
    for parent in args:
        node.add_logical_edge(parent.uid)

    return node

pipeline(fn)

Pipeline decorator.

Parameters:

Name	Type	Description	Default
fn	Callable[..., Any]	Pipeline function.	required

Raises:

Type	Description
TaskNotUniqueError	Another pipeline with the same name has already bee imported.

Returns:

Name	Type	Description
Pipeline	Pipeline	Pipeline.

Source code in python/sdag/wrappers.py

def pipeline(fn: Callable[..., Any]) -> Pipeline:
    """Pipeline decorator.

    Args:
        fn (Callable[..., Any]): Pipeline function.

    Raises:
        TaskNotUniqueError: Another pipeline with
            the same name has already bee imported.

    Returns:
        Pipeline: Pipeline.
    """
    if fn.__name__ in master.pipelines:
        raise TaskNotUniqueError(name=fn.__name__)

    pipeline_obj = Pipeline(fn=fn)
    master.add_pipeline(pipeline_obj)

    return pipeline_obj

task(script, name=None, cmd='sbatch', mode='wrap', cache='none', cache_ignore=None, cache_size=1, njobs=1, retries=0, tags=None)

Local task decorator.

Parameters:

Name	Type	Description	Default
script	str | Path | Script	Launch script. It can be either or a sdag.Script instance.	required
name	str | None	Task name. If null, it will default to the function name. Defaults to None.	None
cmd	Commands	Command to execute the script. Defaults to "sbatch".	'sbatch'
mode	ExecMode	Use wrap to call the Python function or ext to call an external script. Defaults to "wrap".	'wrap'
cache	CacheOptions	Global caching options. 'none': No caching. 'output': All output artifacts must exist. 'io': Input values must match and all output artifacts exist. 'task': Same as 'io' but the task function code is also checked. 'project': Same as 'io' but the code of all internal modules imported when the pipeline is compiled must also match. Defaults to 'none'.	'none'
cache_ignore	list[str] | None	List of input arguments ignored during cache validation. Ignored if cache is set to 'none' or 'output'. Defaults to None.	None
cache_size	int	Cache size. Ignore if caching is disabled. set to 0 to allow for infinite cache size.	1
njobs	int	When spreading a collection over multiple jobs with each(), njobs represents the number of jobs in the array. It is ignored if the task is not an array. Defaults to 1.	1
retries	int	(int): Number of retries. Defaults to 0.	0
tags	list[str] | None	Task tags, they can be used to configure sets of tasks globally.	None

Returns:

Type	Description
Callable[[Callable], Task]	Callable[[Callable], Task]: Task wrapper.

Source code in python/sdag/wrappers.py

def task(
    script: str | Path | Script,
    name: str | None = None,
    cmd: Commands = "sbatch",
    mode: ExecMode = "wrap",
    cache: CacheOptions = "none",
    cache_ignore: list[str] | None = None,
    cache_size: int = 1,
    njobs: int = 1,
    retries: int = 0,
    tags: list[str] | None = None,
) -> Callable[[Callable], Task]:
    """Local task decorator.

    Args:
        script (str | Path | Script): Launch script. It can be
            either or a sdag.Script instance.
        name (str | None, optional): Task name. If null, it will
            default to the function name. Defaults to None.
        cmd (Commands, optional): Command to
            execute the script. Defaults to "sbatch".
        mode (ExecMode, optional): Use wrap to call the
            Python function or ext to call an external script.
            Defaults to "wrap".
        cache (CacheOptions, optional): Global caching options.
            'none': No caching. 'output': All output artifacts
            must exist. 'io': Input values must match and all
            output artifacts exist. 'task': Same as 'io'
            but the task function code is also checked.
            'project': Same as 'io' but the code of all internal
            modules imported when the pipeline is compiled must also
            match. Defaults to 'none'.
        cache_ignore (list[str] | None): List of input arguments
            ignored during cache validation. Ignored if cache is
            set to 'none' or 'output'. Defaults to None.
        cache_size (int): Cache size. Ignore if caching is
            disabled. set to 0 to allow for infinite cache
            size.
        njobs (int): When spreading a collection over multiple
            jobs with each(), njobs represents the number of
            jobs in the array. It is ignored if the task is not
            an array. Defaults to 1.
        retries: (int): Number of retries. Defaults to 0.
        tags (list[str] | None): Task tags, they can be used to
            configure sets of tasks globally.

    Returns:
        Callable[[Callable], Task]: Task wrapper.
    """

    def return_task(fn: Callable[..., Any]) -> Task:
        task_name = name if name is not None else fn.__name__
        if isinstance(script, Script):
            script_obj = ScriptContent(content=script.content)
        else:
            script_obj = ScriptPath(path=Path(script))

        task = Task(
            fn=fn,
            name=task_name,
            scope="global",
            cmd=cmd,
            mode=mode,
            cache=cache,
            cache_ignore=cache_ignore,
            cache_size=cache_size,
            njobs=njobs,
            retries=retries,
            script=script_obj,
            tags=tags if tags is not None else [],
        )
        master.add_task(task)

        return task

    return return_task

validate_kwargs(sig, kwargs)

Validate kwarg consistency.

Parameters:

Name	Type	Description	Default
sig	Signature	Function signature.	required
kwargs	dict[str, Any]	Input kwargs.	required

Raises:

Type	Description
KwargNotFoundError	Keyword argument not found in the input values.

Source code in python/sdag/wrappers.py

def validate_kwargs(sig: inspect.Signature, kwargs: dict[str, Any]) -> None:
    """Validate kwarg consistency.

    Args:
        sig (inspect.Signature): Function signature.
        kwargs (dict[str, Any]): Input kwargs.

    Raises:
        KwargNotFoundError: Keyword argument not found in
            the input values.
    """
    any_var = any(p.kind == p.VAR_KEYWORD for p in sig.parameters.values())
    for key in kwargs:
        if key not in sig.parameters and not any_var:
            raise KwargNotFoundError(key)

    for key, param in sig.parameters.items():
        if key not in kwargs and not param.kind == param.VAR_KEYWORD:
            raise KwargNotFoundError(key)