Core Library (daggerml)#

DaggerML - A Python library for building and managing directed acyclic graphs.

This library provides tools for creating, manipulating, and executing DAGs with strong typing support and a context-manager based interface.

class daggerml.Dag(_dml: daggerml.core.Dml, _message_handler: Callable | None = None, _ref: daggerml.core.Ref | None = None, _init_complete: bool = False)[source]#

Bases: object

Parameters:
  • _dml (Dml)

  • _message_handler (Callable | None)

  • _ref (Ref | None)

  • _init_complete (bool)

property argv: Node#

Access the dag’s argv node

property keys: list[str]#
property result: Node#
property values: list[Node]#
class daggerml.Dml(config_dir=None, project_dir=None, cache_path=None, repo=None, user=None, branch=None, token=None, tmpdirs=<factory>)[source]#

Bases: object

DaggerML cli client wrapper

Parameters:
  • config_dir (str | None)

  • project_dir (str | None)

  • cache_path (str | None)

  • repo (str | None)

  • user (str | None)

  • branch (str | None)

  • token (str | None)

  • tmpdirs (dict[str, TemporaryDirectory])

branch: str | None = None#
cache_path: str | None = None#
cleanup()[source]#
config_dir: str | None = None#
property envvars#
property kwargs: dict#
load(name, recurse=False)[source]#
Parameters:

name (str | Node)

Return type:

Dag

new(name='', message='', data=None, message_handler=None)[source]#
Return type:

Dag

project_dir: str | None = None#
repo: str | None = None#
classmethod temporary(repo='test', user='user', branch='main', cache_path=None, **kwargs)[source]#

Create a temporary Dml instance with specified parameters.

Parameters:
  • repo (str, default="test")

  • user (str, default="user")

  • branch (str, default="main")

  • **kwargs (dict) – Additional keyword arguments for configuration include config_dir, project_dir, and cache_path. If any of those is provided, it will not create a temporary directory for that parameter. If provided and set to None, the dml default will be used.

Return type:

Dml

tmpdirs: dict[str, TemporaryDirectory]#
token: str | None = None#
user: str | None = None#
exception daggerml.Error(message, context=<factory>, code=None)[source]#

Bases: Exception

Custom error type for DaggerML.

Parameters:
  • message (Union[str, Exception]) – Error message or exception

  • context (dict, optional) – Additional error context

  • code (str, optional) – Error code

Return type:

None

code: str | None = None#
context: dict#
message: str | Exception#
class daggerml.Node(dag, ref, _info=<factory>)[source]#

Bases: object

Representation of a node in a DaggerML DAG.

Parameters:
  • dag (Dag) – Parent DAG

  • ref (Ref) – Node reference

  • _info (dict)

property argv: Node#

Access the node’s argv list

dag: Dag#
load(*keys)[source]#

Convenience wrapper around dml.load(node)

If key is provided, it considers this node to be a collection created by the appropriate method and loads the dag that corresponds to this key

Parameters:

*keys (str, optional) – Key to load from the DAG. If not provided, the entire DAG is loaded.

Returns:

The dag that this node was imported from (or in the case of a function call, this returns the fndag)

Return type:

Dag

Examples

>>> dml = Dml.temporary()
>>> dag = dml.new("test", "test")
>>> l0 = dag._put(42)
>>> c0 = dag._put({"a": 1, "b": [l0, "23"]})
>>> assert c0.load("b", 0) == l0
>>> assert c0.load("b").load(0) == l0
>>> assert c0["b"][0] != l0  # this is a different node, not the same as l0
>>> dml.cleanup()
ref: Ref#
property type#

Get the data type of the node.

value()[source]#

Get the concrete value of this node.

Returns:

The actual value represented by this node

Return type:

Any

class daggerml.Resource(uri, data=None, adapter=None)[source]#

Bases: object

Representation of an externally managed object with an identifier.

Parameters:
  • uri (str) – Resource URI

  • data (str, optional) – Associated data

  • adapter (str, optional) – Resource adapter name

adapter: str | None = None#
data: str | None = None#
uri: str#