DaggerML public package exports.
1"""DaggerML public package exports.""" 2 3from daggerml._core import ( 4 BadExecutionStatusError, 5 CanceledExecutionError, 6 CancellationError, 7 Dml, 8 Error, 9 ExecutionDriver, 10 ExecutionMetadata, 11 ExecutionRecord, 12 ExecutionSemanticState, 13 Ref, 14 Runnable, 15 Uri, 16) 17from daggerml.api import ( 18 Dag, 19 Node, 20 clear_default_dml, 21 get_default_dml, 22 load, 23 new, 24 resume, 25 set_default_dml, 26 status, 27 temporary, 28 use_default_dml, 29) 30 31try: 32 from daggerml.__about__ import __version__ 33except ImportError: 34 __version__ = "local" 35 36__all__ = ( 37 "BadExecutionStatusError", 38 "CanceledExecutionError", 39 "CancellationError", 40 "Dag", 41 "Dml", 42 "Error", 43 "ExecutionDriver", 44 "ExecutionMetadata", 45 "ExecutionRecord", 46 "ExecutionSemanticState", 47 "Node", 48 "Ref", 49 "Uri", 50 "Runnable", 51 "get_default_dml", 52 "set_default_dml", 53 "use_default_dml", 54 "clear_default_dml", 55 "new", 56 "load", 57 "resume", 58 "status", 59 "temporary", 60)
484class BadExecutionStatusError(DmlRepoError): 485 """Raised when an execution lifecycle cannot satisfy a requested mutation mode.""" 486 487 def __init__(self, message: str, *, lifecycle: str | None = None): 488 super().__init__(message, type="badexecutionstatuserror") 489 self.lifecycle = lifecycle
Raised when an execution lifecycle cannot satisfy a requested mutation mode.
492class CanceledExecutionError(BadExecutionStatusError): 493 """Raised when cancellation lifecycle blocks activation or mutation.""" 494 495 def __init__(self, message: str, *, lifecycle: str | None = None): 496 super().__init__(message, lifecycle=lifecycle) 497 self.type = "canceledexecutionerror"
Raised when cancellation lifecycle blocks activation or mutation.
Raised when cancellation lifecycle blocks activation or mutation.
340@dataclass 341class Dag: 342 dml: Dml 343 token: Optional[Ref] = None # Working index id 344 ref: Optional[Ref] = None 345 name: str = "" # DAG name for commit 346 message: str = "" # Commit message 347 348 def __repr__(self): 349 to = self.ref.to if self.ref else (self.token.to if self.token is not None else "NA") 350 return f"Dag({to})" 351 352 def __hash__(self): 353 "Useful only for tests." 354 return 42 355 356 def __eq__(self, other): 357 "DAG equality is based on identity, not content." 358 if not isinstance(other, Dag): 359 return False 360 return self.ref == other.ref and self.token == other.token and self.dml == other.dml 361 362 def __enter__(self): 363 "Catch exceptions and commit an Error" 364 assert not self.ref 365 return self 366 367 def __exit__(self, exc_type, exc_value, tb): 368 if exc_value is not None: 369 # Convert exception to Error and commit it 370 traceback.print_exception(exc_type, exc_value, tb) 371 err = Error.from_ex(exc_value) if not isinstance(exc_value, Error) else exc_value 372 self.commit(err) 373 374 def _require_index_ref(self) -> Ref: 375 if self.token is None: 376 raise DmlRepoError("No active index") 377 return self.token 378 379 def _read_dag_ref(self) -> Ref: 380 """Return the completed or partial DAG backing this wrapper's reads.""" 381 if self.ref is not None: 382 return self.ref 383 return cast(Ref, self.dml.runtime.describe(self._require_index_ref())["dag"]) 384 385 @property 386 def tags(self) -> list[str]: 387 """Return the normalized tags stored on this DAG.""" 388 return list(self.dml.dag.describe(self._read_dag_ref())["tags"]) 389 390 def _put_literal(self, value: Any, *, name: Optional[str] = None) -> Ref: 391 index_id = self._require_index_ref() 392 value = apply_codecs(value, dag=self) 393 return self.dml.runtime.put_literal(index_id, value, name=name) 394 395 def _start_fn(self, argv: list[Ref], *, name: Optional[str] = None) -> Optional[Ref]: 396 return self.dml.runtime.start_fn(self._require_index_ref(), argv, name=name) 397 398 def _call_builtin(self, uri: str, *args: Any, name: Optional[str] = None) -> Ref: 399 fn_ref = self._put_literal(Runnable(target=Uri(uri), kwargs={}, adapter="")) 400 argv: list[Ref] = [fn_ref] 401 for arg in args: 402 argv.append(arg if isinstance(arg, Ref) else self._put_literal(arg)) 403 result = self._start_fn(argv, name=name) 404 if result is None: 405 raise DmlRepoError("Function execution failed") 406 return result 407 408 def __len__(self) -> int: 409 return len(self.keys()) 410 411 def __iter__(self): 412 yield from self.keys() 413 414 def _get_named_node(self, name: str) -> "Node": 415 node_ref = self.dml.dag.describe(self._read_dag_ref())["names"].get(name) 416 if node_ref is None: 417 raise DmlRepoError(f"Node '{name}' not found in DAG") 418 return _make_node(self, node_ref) 419 420 def _set_named_node(self, name: str, value: Any) -> None: 421 if self.ref is not None: 422 raise DmlRepoError("Cannot set node names on a committed DAG.") 423 if isinstance(value, Node): 424 value = value.ref 425 if isinstance(value, Ref): 426 self.dml.runtime.set_node_name(self._require_index_ref(), name, value) 427 return 428 self.put(value, name=name) 429 430 def __getitem__(self, name: str) -> "Node": 431 return self._get_named_node(name) 432 433 def __setitem__(self, name: str, value: Any) -> None: 434 self.put(value, name=name) 435 436 def __getattr__(self, name: str) -> "Node": 437 if name.startswith("_"): 438 raise AttributeError(name) 439 return self[name] 440 441 def __setattr__(self, name: str, value: Any) -> None: 442 dataclass_fields = getattr(type(self), "__dataclass_fields__", {}) 443 if name in dataclass_fields or hasattr(type(self), name): 444 object.__setattr__(self, name, value) 445 return 446 self[name] = value 447 448 def keys(self) -> list[str]: 449 """Get the list of all node names in the dag""" 450 names_dict = self.dml.dag.describe(self._read_dag_ref())["names"] 451 return sorted(names_dict.keys()) 452 453 def values(self) -> list["Node"]: 454 """Get the list of all nodes in the dag""" 455 names_dict = self.dml.dag.describe(self._read_dag_ref())["names"] 456 return [_make_node(self, ref) for ref in names_dict.values()] 457 458 @property 459 def argv(self) -> "ListNode": 460 "Access the dag's argv node" 461 argv_ref = self.dml.dag.describe(self._read_dag_ref())["argv"] 462 if not isinstance(argv_ref, Ref): 463 raise DmlRepoError(f"'{self.__class__.__name__}' dag has no argv") 464 return cast(ListNode, _make_node(self, argv_ref)) 465 466 @property 467 def result(self) -> "Node": 468 """Get the result node of the dag""" 469 if self.ref is None: 470 raise DmlRepoError("Cannot access result of an uncommitted DAG") 471 description = self.dml.dag.describe(self.ref) 472 error_ref = description.get("error") 473 if isinstance(error_ref, Ref): 474 raise self.dml.dag.get_error(error_ref) 475 ref = description.get("result") 476 if not isinstance(ref, Ref): 477 raise DmlRepoError(f"'{self.__class__.__name__}' dag has not been committed yet") 478 return _make_node(self, ref) 479 480 @overload 481 def put(self, value: Union[list, "ListNode"], *, name=None) -> "ListNode": ... 482 @overload 483 def put(self, value: Union[dict, "DictNode"], *, name=None) -> "DictNode": ... 484 @overload 485 def put(self, value: Union[Runnable, "RunnableNode"], *, name=None) -> "RunnableNode": ... 486 @overload 487 def put(self, value: Union[Scalar, "ScalarNode"], *, name=None) -> "ScalarNode": ... 488 @overload 489 def put(self, value: Any, *, name=None) -> "Node": ... 490 def put(self, value: Any, *, name=None) -> "Node": 491 """ 492 Add a value to the DAG. 493 494 Parameters 495 ---------- 496 value : Union[Scalar, Collection] 497 Value to add 498 name : str, optional 499 Name for the node 500 Returns 501 ------- 502 Node 503 Node representing the value 504 505 Examples 506 -------- 507 >>> n1 = dag.put(42, name="answer") 508 >>> n1.value() 509 42 510 >>> n2 = dag.put({"a": 1, "b": [n1, "23"]}) 511 >>> n2.value() 512 {'a': 1, 'b': [42, '23']} 513 >>> n3 = dag.put({"a": 1, "b": [n1, "23"]}) 514 >>> n3.value() 515 {'a': 1, 'b': [42, '23']} 516 """ 517 return _make_node(self, self._put_literal(value, name=name)) 518 519 def require(self, dag_name: str | "Dag", node_name: str | None = None, *, name: str | None = None) -> "Node": 520 """ 521 Import a node from a different (committed) DAG into the current DAG. 522 523 Parameters 524 ---------- 525 dag_name : str 526 Name of the DAG to import from 527 node_name : str, optional 528 Name of the node to import. If None, imports the result node of the DAG. 529 530 Returns 531 ------- 532 Node 533 The loaded node or DAG 534 535 Examples 536 -------- 537 >>> dag = new(dml=dml, name="test", message="test") 538 >>> n1 = dag.put(42, name="answer") 539 >>> n2 = dag.put({"a": 1, "b": [n1, "23"]}, name="data") 540 >>> dag.commit(n2) 541 >>> dag2 = new(dml=dml, name="test2", message="test2") 542 >>> imported_n2 = dag2.require("test", "data", name="imported_data") 543 >>> imported_n2.value() 544 {'a': 1, 'b': [42, '23']} 545 """ 546 index = self._require_index_ref() 547 if isinstance(dag_name, Dag): 548 if dag_name.ref is None: 549 raise DmlRepoError("Cannot import an uncommitted DAG") 550 dag_ref = dag_name.ref 551 else: 552 commit = self.dml.runtime.describe(index)["parents"][0] 553 dag_ref = self.dml.show(revision=commit)["dags"].get(dag_name) 554 if dag_ref is None: 555 raise DmlRepoError(f"DAG not found: {dag_name}") 556 dag_info = self.dml.dag.describe(dag_ref) 557 node_ref = dag_info["names"].get(node_name) if node_name else dag_info.get("result") 558 if node_ref is None: 559 raise DmlRepoError(f"Node '{node_name}' not found in DAG '{dag_name}'") 560 node_ref = self.dml.runtime.put_import(index, dag_ref, node_ref, name=name) 561 return _make_node(self, node_ref) 562 563 def call( 564 self, 565 fn: Any, 566 *args: Any, 567 name: Optional[str] = None, 568 sleep: Optional[callable] = None, 569 timeout: int = -1, 570 ) -> "Node": 571 """ 572 Call a function node with arguments. 573 574 Parameters 575 ---------- 576 fn : Union[Runnable, RunnableNode] 577 Function to call 578 *args : Union[Node, Scalar, Collection] 579 Arguments to pass to the function 580 name : str, optional 581 Name for the result node 582 sleep : callable, optional 583 A nullary function that returns sleep time in milliseconds 584 timeout : int, default=-1 585 Maximum time to wait in milliseconds. If <= 0, wait indefinitely. 586 587 Returns 588 ------- 589 Node 590 Result node 591 592 Raises 593 ------ 594 TimeoutError 595 If the function call exceeds the timeout 596 Error 597 If the function returns an error 598 """ 599 sleep = sleep or BackoffWithJitter() 600 argv_seed = [fn, *args] 601 end = current_time_millis() + timeout 602 while timeout <= 0 or current_time_millis() < end: 603 argv_refs = [self._put_literal(value) for value in argv_seed] 604 resp = self._start_fn(argv_refs, name=name) 605 if resp: 606 return _make_node(self, resp) 607 time.sleep(sleep() / 1000) 608 raise TimeoutError(f"invoking function: {fn}") 609 610 def commit(self, value) -> None: 611 """ 612 Commit a value to the DAG. 613 614 Parameters 615 ---------- 616 value : Union[Node, Error, Any] 617 Value to commit 618 """ 619 # errors are committed as-is, everything else is a node 620 if not isinstance(value, (Error, Node)): 621 value = self.put(value) 622 if isinstance(value, Node): 623 value = value.ref 624 self.ref = self.dml.runtime.commit(self._require_index_ref(), value, message=self.message, name=self.name) 625 self.token = None # Clear the working index since it's now committed 626 627 def freeze(self, message: str | None = None) -> "Dag": 628 """Freeze this uncommitted DAG's runtime index for read-only inspection.""" 629 freeze_message = f"dag: {self.name}" 630 if message: 631 freeze_message = f"{freeze_message}\n{message}" 632 self.token = self.dml.runtime.freeze(self._require_index_ref(), message=freeze_message) 633 return self 634 635 def unfreeze(self) -> "Dag": 636 """Unfreeze this uncommitted DAG's runtime index.""" 637 self.token = self.dml.runtime.unfreeze(self._require_index_ref()) 638 return self 639 640 def cancel(self, max_retries: int = 3): 641 """Cancel the DAG's execution. 642 643 Parameters 644 ---------- 645 max_retries : int, default=3 646 Maximum cancellation retries after the initial attempt. 647 """ 648 if self.token is None: 649 raise DmlRepoError("Cannot cancel a committed DAG") 650 logger.info(f"Cancelling execution {self.token}") 651 self.dml.runtime.cancel(self.token, max_retries=max_retries) 652 self.token = None # Clear the index ref to indicate it's no longer active 653 raise CancellationError("DAG execution cancelled")
458 @property 459 def argv(self) -> "ListNode": 460 "Access the dag's argv node" 461 argv_ref = self.dml.dag.describe(self._read_dag_ref())["argv"] 462 if not isinstance(argv_ref, Ref): 463 raise DmlRepoError(f"'{self.__class__.__name__}' dag has no argv") 464 return cast(ListNode, _make_node(self, argv_ref))
Access the dag's argv node
466 @property 467 def result(self) -> "Node": 468 """Get the result node of the dag""" 469 if self.ref is None: 470 raise DmlRepoError("Cannot access result of an uncommitted DAG") 471 description = self.dml.dag.describe(self.ref) 472 error_ref = description.get("error") 473 if isinstance(error_ref, Ref): 474 raise self.dml.dag.get_error(error_ref) 475 ref = description.get("result") 476 if not isinstance(ref, Ref): 477 raise DmlRepoError(f"'{self.__class__.__name__}' dag has not been committed yet") 478 return _make_node(self, ref)
Get the result node of the dag
490 def put(self, value: Any, *, name=None) -> "Node": 491 """ 492 Add a value to the DAG. 493 494 Parameters 495 ---------- 496 value : Union[Scalar, Collection] 497 Value to add 498 name : str, optional 499 Name for the node 500 Returns 501 ------- 502 Node 503 Node representing the value 504 505 Examples 506 -------- 507 >>> n1 = dag.put(42, name="answer") 508 >>> n1.value() 509 42 510 >>> n2 = dag.put({"a": 1, "b": [n1, "23"]}) 511 >>> n2.value() 512 {'a': 1, 'b': [42, '23']} 513 >>> n3 = dag.put({"a": 1, "b": [n1, "23"]}) 514 >>> n3.value() 515 {'a': 1, 'b': [42, '23']} 516 """ 517 return _make_node(self, self._put_literal(value, name=name))
Add a value to the DAG.
>>> n1 = dag.put(42, name="answer")
>>> n1.value()
42
>>> n2 = dag.put({"a": 1, "b": [n1, "23"]})
>>> n2.value()
{'a': 1, 'b': [42, '23']}
>>> n3 = dag.put({"a": 1, "b": [n1, "23"]})
>>> n3.value()
{'a': 1, 'b': [42, '23']}
519 def require(self, dag_name: str | "Dag", node_name: str | None = None, *, name: str | None = None) -> "Node": 520 """ 521 Import a node from a different (committed) DAG into the current DAG. 522 523 Parameters 524 ---------- 525 dag_name : str 526 Name of the DAG to import from 527 node_name : str, optional 528 Name of the node to import. If None, imports the result node of the DAG. 529 530 Returns 531 ------- 532 Node 533 The loaded node or DAG 534 535 Examples 536 -------- 537 >>> dag = new(dml=dml, name="test", message="test") 538 >>> n1 = dag.put(42, name="answer") 539 >>> n2 = dag.put({"a": 1, "b": [n1, "23"]}, name="data") 540 >>> dag.commit(n2) 541 >>> dag2 = new(dml=dml, name="test2", message="test2") 542 >>> imported_n2 = dag2.require("test", "data", name="imported_data") 543 >>> imported_n2.value() 544 {'a': 1, 'b': [42, '23']} 545 """ 546 index = self._require_index_ref() 547 if isinstance(dag_name, Dag): 548 if dag_name.ref is None: 549 raise DmlRepoError("Cannot import an uncommitted DAG") 550 dag_ref = dag_name.ref 551 else: 552 commit = self.dml.runtime.describe(index)["parents"][0] 553 dag_ref = self.dml.show(revision=commit)["dags"].get(dag_name) 554 if dag_ref is None: 555 raise DmlRepoError(f"DAG not found: {dag_name}") 556 dag_info = self.dml.dag.describe(dag_ref) 557 node_ref = dag_info["names"].get(node_name) if node_name else dag_info.get("result") 558 if node_ref is None: 559 raise DmlRepoError(f"Node '{node_name}' not found in DAG '{dag_name}'") 560 node_ref = self.dml.runtime.put_import(index, dag_ref, node_ref, name=name) 561 return _make_node(self, node_ref)
Import a node from a different (committed) DAG into the current DAG.
>>> dag = new(dml=dml, name="test", message="test")
>>> n1 = dag.put(42, name="answer")
>>> n2 = dag.put({"a": 1, "b": [n1, "23"]}, name="data")
>>> dag.commit(n2)
>>> dag2 = new(dml=dml, name="test2", message="test2")
>>> imported_n2 = dag2.require("test", "data", name="imported_data")
>>> imported_n2.value()
{'a': 1, 'b': [42, '23']}
563 def call( 564 self, 565 fn: Any, 566 *args: Any, 567 name: Optional[str] = None, 568 sleep: Optional[callable] = None, 569 timeout: int = -1, 570 ) -> "Node": 571 """ 572 Call a function node with arguments. 573 574 Parameters 575 ---------- 576 fn : Union[Runnable, RunnableNode] 577 Function to call 578 *args : Union[Node, Scalar, Collection] 579 Arguments to pass to the function 580 name : str, optional 581 Name for the result node 582 sleep : callable, optional 583 A nullary function that returns sleep time in milliseconds 584 timeout : int, default=-1 585 Maximum time to wait in milliseconds. If <= 0, wait indefinitely. 586 587 Returns 588 ------- 589 Node 590 Result node 591 592 Raises 593 ------ 594 TimeoutError 595 If the function call exceeds the timeout 596 Error 597 If the function returns an error 598 """ 599 sleep = sleep or BackoffWithJitter() 600 argv_seed = [fn, *args] 601 end = current_time_millis() + timeout 602 while timeout <= 0 or current_time_millis() < end: 603 argv_refs = [self._put_literal(value) for value in argv_seed] 604 resp = self._start_fn(argv_refs, name=name) 605 if resp: 606 return _make_node(self, resp) 607 time.sleep(sleep() / 1000) 608 raise TimeoutError(f"invoking function: {fn}")
Call a function node with arguments.
610 def commit(self, value) -> None: 611 """ 612 Commit a value to the DAG. 613 614 Parameters 615 ---------- 616 value : Union[Node, Error, Any] 617 Value to commit 618 """ 619 # errors are committed as-is, everything else is a node 620 if not isinstance(value, (Error, Node)): 621 value = self.put(value) 622 if isinstance(value, Node): 623 value = value.ref 624 self.ref = self.dml.runtime.commit(self._require_index_ref(), value, message=self.message, name=self.name) 625 self.token = None # Clear the working index since it's now committed
Commit a value to the DAG.
627 def freeze(self, message: str | None = None) -> "Dag": 628 """Freeze this uncommitted DAG's runtime index for read-only inspection.""" 629 freeze_message = f"dag: {self.name}" 630 if message: 631 freeze_message = f"{freeze_message}\n{message}" 632 self.token = self.dml.runtime.freeze(self._require_index_ref(), message=freeze_message) 633 return self
Freeze this uncommitted DAG's runtime index for read-only inspection.
640 def cancel(self, max_retries: int = 3): 641 """Cancel the DAG's execution. 642 643 Parameters 644 ---------- 645 max_retries : int, default=3 646 Maximum cancellation retries after the initial attempt. 647 """ 648 if self.token is None: 649 raise DmlRepoError("Cannot cancel a committed DAG") 650 logger.info(f"Cancelling execution {self.token}") 651 self.dml.runtime.cancel(self.token, max_retries=max_retries) 652 self.token = None # Clear the index ref to indicate it's no longer active 653 raise CancellationError("DAG execution cancelled")
Cancel the DAG's execution.
1068class Dml: 1069 def _init_from_config_vars(self, explicit_config: Mapping[str, object]) -> None: 1070 self._explicit_config = dict(explicit_config) 1071 self._config = Config.resolve(explicit=self._explicit_config) 1072 dflt = self._config.default 1073 self._db = DmlDB(self._config.db_path, dflt.db_map_size_headroom, dflt.db_map_size_max) 1074 self._s3_client = None 1075 1076 def __init__( 1077 self, 1078 project_home: Annotated[str | None, "Project root containing the .dml repository."] = None, 1079 *, 1080 db_path: Annotated[str | None, "Override path to the LMDB database."] = None, 1081 db_map_size_headroom: Annotated[int | None, "Extra LMDB map size headroom in bytes."] = None, 1082 db_map_size_max: Annotated[int | None, "Maximum LMDB map size in bytes."] = None, 1083 default_branch_name: Annotated[str | None, "Default branch name for attached HEAD operations."] = None, 1084 remote_root: Annotated[str | None, "Remote storage root URI."] = None, 1085 remote_prune_age_seconds: Annotated[int | None, "Remote GC prune age in seconds."] = None, 1086 remote_fetch_workers: Annotated[int | None, "Number of concurrent remote fetch workers."] = None, 1087 user: Annotated[str | None, "User name recorded in commits and runtime actions."] = None, 1088 config_home: Annotated[str | None, "Override config directory path."] = None, 1089 ): 1090 """Create a DaggerML session bound to one repository and config context.""" 1091 self._init_from_config_vars( 1092 _python_config_vars_to_canonical( 1093 project_home=project_home, 1094 db_path=db_path, 1095 db_map_size_headroom=db_map_size_headroom, 1096 db_map_size_max=db_map_size_max, 1097 default_branch_name=default_branch_name, 1098 remote_root=remote_root, 1099 remote_prune_age_seconds=remote_prune_age_seconds, 1100 remote_fetch_workers=remote_fetch_workers, 1101 user=user, 1102 config_home=config_home, 1103 ) 1104 ) 1105 1106 @classmethod 1107 def from_config_vars( 1108 cls, 1109 config_vars: Annotated[dict[str, object], "Flattened canonical config-var mapping."] | None = None, 1110 ) -> "Dml": 1111 """Create a DaggerML session from flattened canonical config vars.""" 1112 dml = cls.__new__(cls) 1113 dml._init_from_config_vars(config_vars or {}) 1114 return dml 1115 1116 @classmethod 1117 def init( 1118 cls, 1119 project_home: Annotated[str, "Directory where the repository should be initialized."] = ".", 1120 *, 1121 db_path: Annotated[str | None, "Override path to the LMDB database."] = None, 1122 db_map_size_headroom: Annotated[int | None, "Extra LMDB map size headroom in bytes."] = None, 1123 db_map_size_max: Annotated[int | None, "Maximum LMDB map size in bytes."] = None, 1124 default_branch_name: Annotated[str | None, "Default branch name for attached HEAD operations."] = None, 1125 remote_root: Annotated[str | None, "Remote storage root URI."] = None, 1126 remote_prune_age_seconds: Annotated[int | None, "Remote GC prune age in seconds."] = None, 1127 remote_fetch_workers: Annotated[int | None, "Number of concurrent remote fetch workers."] = None, 1128 user: Annotated[str | None, "User name recorded in commits and runtime actions."] = None, 1129 config_home: Annotated[str | None, "Override config directory path."] = None, 1130 branch: Annotated[str | None, "Initial branch name."] = None, 1131 ) -> "Dml": 1132 """Initialize a repository with an unborn attached HEAD.""" 1133 config = Config.init(project_home, remote_root=remote_root) 1134 dml = cls.from_config_vars( 1135 _python_config_vars_to_canonical( 1136 project_home=config.project_home, 1137 db_path=db_path, 1138 db_map_size_headroom=db_map_size_headroom, 1139 db_map_size_max=db_map_size_max, 1140 default_branch_name=default_branch_name, 1141 remote_root=remote_root, 1142 remote_prune_age_seconds=remote_prune_age_seconds, 1143 remote_fetch_workers=remote_fetch_workers, 1144 user=user, 1145 config_home=config_home, 1146 ) 1147 ) 1148 head = Head(config.project_home) 1149 branch = branch or dml._config.default.branch_name 1150 with head.lock(): 1151 try: 1152 head.get_head() 1153 except (DmlRepoError, FileNotFoundError): 1154 dml._db.init() 1155 head.init(None, branch) 1156 return dml 1157 1158 @classmethod 1159 def clone( 1160 cls, 1161 revision: Annotated[Ref | str | None, "Optional branch, tag, or commit revision."] = None, 1162 /, 1163 *, 1164 project_home: Annotated[str, "Directory where the repository should be cloned."] = ".", 1165 db_path: Annotated[str | None, "Override path to the LMDB database."] = None, 1166 db_map_size_headroom: Annotated[int | None, "Extra LMDB map size headroom in bytes."] = None, 1167 db_map_size_max: Annotated[int | None, "Maximum LMDB map size in bytes."] = None, 1168 default_branch_name: Annotated[str | None, "Default branch name for attached HEAD operations."] = None, 1169 remote_root: Annotated[str | None, "Remote storage root URI."] = None, 1170 remote_prune_age_seconds: Annotated[int | None, "Remote GC prune age in seconds."] = None, 1171 remote_fetch_workers: Annotated[int | None, "Number of concurrent remote fetch workers."] = None, 1172 user: Annotated[str | None, "User name recorded in commits and runtime actions."] = None, 1173 config_home: Annotated[str | None, "Override config directory path."] = None, 1174 depth: Annotated[int | None, "Positive number of commit-history generations to fetch."] = None, 1175 ) -> "Dml": 1176 """Clone one revision from the configured remote root.""" 1177 _validate_history_options(depth) 1178 Path(project_home).mkdir(parents=True, exist_ok=True) 1179 config = Config.init(project_home, remote_root=remote_root) 1180 dml = cls.from_config_vars( 1181 _python_config_vars_to_canonical( 1182 project_home=config.project_home, 1183 db_path=db_path, 1184 db_map_size_headroom=db_map_size_headroom, 1185 db_map_size_max=db_map_size_max, 1186 default_branch_name=default_branch_name, 1187 remote_root=remote_root, 1188 remote_prune_age_seconds=remote_prune_age_seconds, 1189 remote_fetch_workers=remote_fetch_workers, 1190 user=user, 1191 config_home=config_home, 1192 ) 1193 ) 1194 _require_remote_root(dml) 1195 selected = revision or dml._config.default.branch_name 1196 exact = selected if isinstance(selected, Ref) else None 1197 if isinstance(selected, str) and re.match(r"^(?:commit:)?[0-9a-f]{64}$", selected): 1198 exact = Ref(selected if selected.startswith("commit:") else f"commit:{selected}") 1199 branch = ( 1200 selected 1201 if exact is None and isinstance(selected, str) and not selected.startswith(("@", "HEAD")) 1202 else None 1203 ) 1204 initial_branch = branch or dml._config.default.branch_name 1205 head = Head(config.project_home) 1206 with head.lock(): 1207 try: 1208 head.get_head() 1209 except (DmlRepoError, FileNotFoundError): 1210 dml._db.init() 1211 head.init(None, initial_branch) 1212 else: 1213 raise DmlRepoError(f"Cannot clone into an initialized repository: {dml._config.project_home}") 1214 if exact is not None: 1215 commit, available, omitted = _remote_ops(dml).materialize_project_commit_ref(exact, dml._db, depth=depth) 1216 with head.lock(): 1217 _publish_shallow_state(head, available, omitted) 1218 head.write_detached_head(commit) 1219 return dml 1220 dml.fetch(selected if isinstance(selected, str) else None, depth=depth) 1221 kind = "tag" if isinstance(selected, str) and selected.startswith("@") else "branch" 1222 name = selected[1:] if kind == "tag" and isinstance(selected, str) else selected 1223 assert isinstance(name, str) 1224 with head.lock(): 1225 commit = head.get_remote_tracking_ref(name, kind=kind) 1226 if kind == "tag": 1227 head.write_detached_head(commit) 1228 else: 1229 head.update_local_ref(name, commit, kind="branch") 1230 head.write_attached_head(name) 1231 head.set_upstream(name, name) 1232 return dml 1233 1234 def status(self) -> StatusPayload: 1235 """Return branch, commit, and open-runtime status for this repository.""" 1236 head = _head_ops(self) 1237 head_info = head.get_head() 1238 ahead = behind = None 1239 upstream = head.get_upstream(head_info["branch"]) if head_info["branch"] is not None else None 1240 if upstream is not None: 1241 try: 1242 upstream_ref = head.get_remote_tracking_ref(upstream["branch"]) 1243 except DmlRepoError: 1244 pass 1245 else: 1246 if head_info["commit"] is not None: 1247 try: 1248 ahead, behind = CommitOps().ahead_behind( 1249 head_info["commit"], 1250 upstream_ref, 1251 db=self._db, 1252 missing_commits=head.get_shallow_commits(), 1253 ) 1254 except ShallowHistoryError: 1255 pass 1256 with self._db.tx(readonly=True) as txn: 1257 num_indexes = 0 1258 for namespace in ("index", "frozenindex"): 1259 try: 1260 num_indexes += sum(1 for _ in txn.iter(namespace)) 1261 except DmlDbKeyNotFoundError: 1262 pass 1263 return { 1264 "mode": head_info["mode"], 1265 "branch": head_info["branch"], 1266 "commit": head_info["commit"], 1267 "branches": head.list_local_refs(kind="branch"), 1268 "upstream": upstream["branch"] if upstream is not None else None, 1269 "num_indexes": num_indexes, 1270 "ahead": ahead, 1271 "behind": behind, 1272 } 1273 1274 def log( 1275 self, 1276 revision: Annotated[Ref | str, "Revision to start the log from."] = "HEAD", 1277 limit: Annotated[int, "Maximum number of commits to return."] = 10, 1278 *, 1279 remote: Annotated[bool, "Resolve from fetched remote tracking refs."] = False, 1280 dep: Annotated[str | None, "Resolve from a fetched dependency."] = None, 1281 ) -> LogPayload: 1282 """Return commit history starting from one revision.""" 1283 commit_ref = resolve_rev(_head_ops(self), revision, db=self._db, remote=remote, dep=dep) 1284 commit_ref = _require_resolved_commit(commit_ref, revision) 1285 commits, truncated = CommitOps().log_with_truncation( 1286 commit_ref, 1287 limit=limit, 1288 db=self._db, 1289 missing_commits=_head_ops(self).get_shallow_commits(), 1290 ) 1291 return {"commits": commits, "truncated": truncated} 1292 1293 def show( 1294 self, 1295 revision: Annotated[Ref | str, "Revision to describe."] = "HEAD", 1296 *, 1297 remote: bool = False, 1298 dep: str | None = None, 1299 ) -> CommitFullDescription: 1300 """Return a full commit description for one revision.""" 1301 commit_ref = resolve_rev(_head_ops(self), revision, db=self._db, remote=remote, dep=dep) 1302 commit_ref = _require_resolved_commit(commit_ref, revision) 1303 return CommitOps().show( 1304 commit_ref, 1305 db=self._db, 1306 missing_commits=_head_ops(self).get_shallow_commits(), 1307 ) 1308 1309 def diff( 1310 self, 1311 revision: Annotated[Ref | str, "Revision to diff."] = "HEAD", 1312 relative_to: Annotated[Ref | str | None, "Optional base revision. Defaults to the commit parent."] = None, 1313 *, 1314 remote: Annotated[bool, "Resolve the primary revision from remote tracking."] = False, 1315 dep: Annotated[str | None, "Resolve the primary revision from a dependency."] = None, 1316 ) -> CommitDiffPayload: 1317 """Return DAG-level changes for one revision.""" 1318 head = _head_ops(self) 1319 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote, dep=dep) 1320 commit_ref = _require_resolved_commit(commit_ref, revision) 1321 if relative_to is None: 1322 return CommitOps().diff( 1323 commit_ref, 1324 db=self._db, 1325 missing_commits=head.get_shallow_commits(), 1326 ) 1327 rel_to_commit = resolve_rev(head, relative_to, db=self._db) 1328 rel_to_commit = _require_resolved_commit(rel_to_commit, relative_to) 1329 return CommitOps().diff( 1330 commit_ref, 1331 rel_to_commit, 1332 db=self._db, 1333 missing_commits=head.get_shallow_commits(), 1334 ) 1335 1336 def rev_parse( 1337 self, 1338 revision: Annotated[str, "Revision expression to resolve."], 1339 *, 1340 remote: bool = False, 1341 dep: str | None = None, 1342 ) -> RevisionPayload: 1343 """Resolve a revision expression into a commit and ref metadata.""" 1344 head = _head_ops(self) 1345 branch = tag = None 1346 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote, dep=dep) 1347 if revision.startswith("HEAD"): 1348 kind = "head" 1349 elif re.match(r"^(?:commit:)?[0-9a-f]{64}$", revision): 1350 kind = "commit" 1351 else: 1352 kind = "ref" 1353 if kind == "ref": 1354 branch, tag = (None, revision[1:]) if revision.startswith("@") else (revision, None) 1355 return { 1356 "input": revision, 1357 "uri": None, 1358 "kind": kind, 1359 "commit": commit_ref, 1360 "branch": branch, 1361 "tag": tag, 1362 } 1363 1364 def revert( 1365 self, 1366 revision: Annotated[Ref | str, "Revision whose changes should be reverted."], 1367 message: Annotated[str | None, "Optional commit message for the revert commit."] = None, 1368 *, 1369 remote: bool = False, 1370 ) -> StatusPayload: 1371 """Revert the changes introduced by one revision.""" 1372 head = _head_ops(self) 1373 with head.lock(): 1374 head_info = head.get_head() 1375 if head_info["branch"] is None: 1376 raise DmlRepoError("Cannot revert when HEAD is detached") 1377 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote) 1378 commit_ref = _require_resolved_commit(commit_ref, revision) 1379 new_commit = CommitOps().revert( 1380 commit_ref, 1381 _require_resolved_commit(head_info["commit"], "HEAD"), 1382 user=self._config.user, 1383 message=message, 1384 db=self._db, 1385 missing_commits=head.get_shallow_commits(), 1386 ) 1387 head.update_local_ref(head_info["branch"], new_commit, kind="branch") 1388 return self.status() 1389 1390 def checkout( 1391 self, revision: Annotated[Ref | str, "Revision to check out."], *, remote: bool = False 1392 ) -> StatusPayload: 1393 """Check out a different revision. 1394 1395 If the revision resolves to a local branch, HEAD stays attached to that branch. 1396 Other revisions detach HEAD at the resolved commit. 1397 """ 1398 head = _head_ops(self) 1399 with head.lock(): 1400 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote) 1401 if ( 1402 not remote 1403 and isinstance(revision, str) 1404 and not revision.startswith(("@", "HEAD")) 1405 and not re.match(r"^(?:commit:)?[0-9a-f]{64}$", revision) 1406 ): 1407 head.write_attached_head(revision) 1408 else: 1409 head.write_detached_head(_require_resolved_commit(commit_ref, revision)) 1410 return self.status() 1411 1412 def merge( 1413 self, 1414 revision: Annotated[Ref | str, "Revision to merge into the current branch."], 1415 ff_only: Annotated[bool, "Whether to only allow fast-forward merges."] = True, 1416 *, 1417 remote: bool = False, 1418 ) -> StatusPayload: 1419 """Merge a revision into the current HEAD.""" 1420 head = _head_ops(self) 1421 with head.lock(): 1422 head_info = head.get_head() 1423 if head_info["branch"] is None: 1424 raise DmlRepoError("Cannot merge when HEAD is detached") 1425 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote) 1426 commit_ref = _require_resolved_commit(commit_ref, revision) 1427 new_commit = CommitOps().merge( 1428 head_info["commit"], 1429 commit_ref, 1430 user=self._config.user, 1431 ff_only=ff_only, 1432 db=self._db, 1433 missing_commits=head.get_shallow_commits(), 1434 ) 1435 head.update_local_ref(head_info["branch"], new_commit) 1436 return self.status() 1437 1438 def rebase( 1439 self, revision: Annotated[Ref | str, "Revision to rebase the current branch onto."], *, remote: bool = False 1440 ) -> StatusPayload: 1441 """Rebase the current HEAD onto a different revision.""" 1442 head = _head_ops(self) 1443 with head.lock(): 1444 head_info = head.get_head() 1445 if head_info["branch"] is None: 1446 raise DmlRepoError("Cannot rebase when HEAD is detached") 1447 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote) 1448 new_commit = CommitOps().rebase( 1449 _require_resolved_commit(head_info["commit"], "HEAD"), 1450 _require_resolved_commit(commit_ref, revision), 1451 user=self._config.user, 1452 db=self._db, 1453 missing_commits=head.get_shallow_commits(), 1454 ) 1455 head.update_local_ref(head_info["branch"], new_commit) 1456 return self.status() 1457 1458 def fetch( 1459 self, 1460 revision: Annotated[str | None, "Branch or @tag to fetch."] = None, 1461 /, 1462 *, 1463 dep: Annotated[str | None, "Named dependency endpoint to fetch from."] = None, 1464 depth: Annotated[int | None, "Positive number of commit-history generations to fetch."] = None, 1465 unshallow: Annotated[bool, "Fetch all history through existing shallow boundaries."] = False, 1466 ) -> None: 1467 """Fetch one branch or tag from remote.root or a named dependency.""" 1468 _validate_history_options(depth, unshallow) 1469 selector = revision or self._config.default.branch_name 1470 kind = "tag" if selector.startswith("@") else "branch" 1471 name = selector[1:] if kind == "tag" else selector 1472 head = _head_ops(self) 1473 if dep is None: 1474 remote_ops = _remote_ops(self) 1475 else: 1476 config = head.get_dependency_config(dep) 1477 remote_ops = Remote( 1478 config["root"], 1479 n_workers=self._config.remote.fetch_workers, 1480 client=_require_s3_client(self), 1481 prune_age_seconds=self._config.remote.prune_age_seconds, 1482 ) 1483 materialized = remote_ops.get_project_commit_ref( 1484 kind, 1485 name, 1486 db=self._db, 1487 depth=depth, 1488 unshallow=unshallow, 1489 ) 1490 if materialized is None: 1491 raise DmlRepoError(f"Remote {kind} ref not found: {selector}") 1492 commit, available, omitted = materialized 1493 with head.lock(): 1494 _publish_shallow_state(head, available, omitted) 1495 if dep is None: 1496 head.update_remote_tracking_ref(name, commit, kind=kind) 1497 else: 1498 head.update_dependency_ref(dep, name, commit, kind=kind) 1499 1500 def pull( 1501 self, 1502 ff_only: Annotated[bool, "Whether to only allow fast-forward merges."] = True, 1503 *, 1504 depth: Annotated[int | None, "Positive number of commit-history generations to fetch."] = None, 1505 ) -> StatusPayload: 1506 """Fetch and merge the current branch's configured upstream.""" 1507 _validate_history_options(depth) 1508 head = _head_ops(self) 1509 head_info = head.get_head() 1510 if head_info["branch"] is None: 1511 raise DmlRepoError("Cannot pull when HEAD is detached") 1512 upstream = head.get_upstream(head_info["branch"]) 1513 if upstream is None: 1514 raise DmlRepoError(f"Cannot pull untracked branch: {head_info['branch']}") 1515 self.fetch(upstream["branch"], depth=depth) 1516 self.merge(upstream["branch"], ff_only=ff_only, remote=True) 1517 return self.status() 1518 1519 def push( 1520 self, 1521 *, 1522 revision: Annotated[Ref | str, "Local revision to publish; defaults to HEAD."] = "HEAD", 1523 force: Annotated[bool, "Overwrite a remote branch or tag without publication checks."] = False, 1524 ) -> None: 1525 """Publish a local revision as a tag or branch on the configured remote.""" 1526 head = _head_ops(self) 1527 commit_ref = _require_resolved_commit(resolve_rev(head, revision, db=self._db), revision) 1528 if isinstance(revision, str) and revision.startswith("@"): 1529 _remote_ops(self).put_ref( 1530 commit_ref, 1531 kind="tag", 1532 name=revision[1:], 1533 db=self._db, 1534 force=force, 1535 missing_commits=head.get_shallow_commits(), 1536 ) 1537 return 1538 named_branch = ( 1539 isinstance(revision, str) 1540 and not revision.startswith("HEAD") 1541 and re.fullmatch(r"(?:commit:)?[0-9a-f]{64}", revision) is None 1542 ) 1543 branch = revision if named_branch and isinstance(revision, str) else head.get_head()["branch"] 1544 if branch is None: 1545 raise DmlRepoError("Cannot push an unnamed revision when HEAD is detached") 1546 upstream = head.get_upstream(branch) 1547 if upstream is None: 1548 upstream = {"branch": branch} 1549 set_upstream = True 1550 else: 1551 set_upstream = False 1552 remote = _remote_ops(self) 1553 remote.put_ref( 1554 commit_ref, 1555 kind="branch", 1556 name=upstream["branch"], 1557 db=self._db, 1558 force=force, 1559 missing_commits=head.get_shallow_commits(), 1560 ) 1561 if set_upstream: 1562 with head.lock(): 1563 head.set_upstream(branch, upstream["branch"]) 1564 1565 def gc( 1566 self, 1567 *, 1568 remote: Annotated[bool, "Garbage-collect configured remote state instead of local objects."] = False, 1569 unrecognized: Annotated[ 1570 Literal["raise", "ignore"], 1571 "Remote only: fail on (raise) or log and skip (ignore) unrecognized objects in immutable storage.", 1572 ] = "raise", 1573 ) -> LocalGCSummary | RemoteGCSummary: 1574 """Garbage-collect unreachable local or configured remote state.""" 1575 if not remote: 1576 return _local_gc(self) 1577 return cast(RemoteGCSummary, _remote_ops(self).gc(unrecognized)) 1578 1579 @property 1580 def branch(self) -> Annotated[_BranchNamespace, "Branch inspection and lifecycle commands."]: 1581 """Expose branch inspection and lifecycle commands.""" 1582 return _BranchNamespace(self) 1583 1584 @property 1585 def dep(self) -> Annotated[_DependencyNamespace, "Import-only dependency lifecycle commands."]: 1586 """Expose import-only dependency lifecycle commands.""" 1587 return _DependencyNamespace(self) 1588 1589 @property 1590 def cache(self) -> Annotated[_CacheNamespace, "Remote execution cache inspection and control commands."]: 1591 """Expose remote execution cache inspection and control commands.""" 1592 return _CacheNamespace(self) 1593 1594 @property 1595 def tag(self) -> Annotated[_TagNamespace, "Tag inspection and lifecycle commands."]: 1596 """Expose tag inspection and lifecycle commands.""" 1597 return _TagNamespace(self) 1598 1599 @property 1600 def config(self) -> Annotated[_ConfigNamespace, "Configuration commands."]: 1601 """Expose configuration commands.""" 1602 return _ConfigNamespace(self) 1603 1604 @property 1605 def runtime(self) -> Annotated[_RuntimeNamespace, "Runtime mutation and execution-state inspection commands."]: 1606 """Expose runtime mutation and execution-state inspection commands.""" 1607 return _RuntimeNamespace(self) 1608 1609 @property 1610 def dag(self) -> Annotated[_DagNamespace, "Committed DAG inspection commands."]: 1611 """Expose committed DAG inspection commands.""" 1612 return _DagNamespace(self) 1613 1614 @property 1615 def skills(self) -> Annotated[_SkillsNamespace, "Bundled agent-guidance exports."]: 1616 """Expose bundled agent-guidance exports.""" 1617 return _SkillsNamespace(self)
1076 def __init__( 1077 self, 1078 project_home: Annotated[str | None, "Project root containing the .dml repository."] = None, 1079 *, 1080 db_path: Annotated[str | None, "Override path to the LMDB database."] = None, 1081 db_map_size_headroom: Annotated[int | None, "Extra LMDB map size headroom in bytes."] = None, 1082 db_map_size_max: Annotated[int | None, "Maximum LMDB map size in bytes."] = None, 1083 default_branch_name: Annotated[str | None, "Default branch name for attached HEAD operations."] = None, 1084 remote_root: Annotated[str | None, "Remote storage root URI."] = None, 1085 remote_prune_age_seconds: Annotated[int | None, "Remote GC prune age in seconds."] = None, 1086 remote_fetch_workers: Annotated[int | None, "Number of concurrent remote fetch workers."] = None, 1087 user: Annotated[str | None, "User name recorded in commits and runtime actions."] = None, 1088 config_home: Annotated[str | None, "Override config directory path."] = None, 1089 ): 1090 """Create a DaggerML session bound to one repository and config context.""" 1091 self._init_from_config_vars( 1092 _python_config_vars_to_canonical( 1093 project_home=project_home, 1094 db_path=db_path, 1095 db_map_size_headroom=db_map_size_headroom, 1096 db_map_size_max=db_map_size_max, 1097 default_branch_name=default_branch_name, 1098 remote_root=remote_root, 1099 remote_prune_age_seconds=remote_prune_age_seconds, 1100 remote_fetch_workers=remote_fetch_workers, 1101 user=user, 1102 config_home=config_home, 1103 ) 1104 )
Create a DaggerML session bound to one repository and config context.
1106 @classmethod 1107 def from_config_vars( 1108 cls, 1109 config_vars: Annotated[dict[str, object], "Flattened canonical config-var mapping."] | None = None, 1110 ) -> "Dml": 1111 """Create a DaggerML session from flattened canonical config vars.""" 1112 dml = cls.__new__(cls) 1113 dml._init_from_config_vars(config_vars or {}) 1114 return dml
Create a DaggerML session from flattened canonical config vars.
1116 @classmethod 1117 def init( 1118 cls, 1119 project_home: Annotated[str, "Directory where the repository should be initialized."] = ".", 1120 *, 1121 db_path: Annotated[str | None, "Override path to the LMDB database."] = None, 1122 db_map_size_headroom: Annotated[int | None, "Extra LMDB map size headroom in bytes."] = None, 1123 db_map_size_max: Annotated[int | None, "Maximum LMDB map size in bytes."] = None, 1124 default_branch_name: Annotated[str | None, "Default branch name for attached HEAD operations."] = None, 1125 remote_root: Annotated[str | None, "Remote storage root URI."] = None, 1126 remote_prune_age_seconds: Annotated[int | None, "Remote GC prune age in seconds."] = None, 1127 remote_fetch_workers: Annotated[int | None, "Number of concurrent remote fetch workers."] = None, 1128 user: Annotated[str | None, "User name recorded in commits and runtime actions."] = None, 1129 config_home: Annotated[str | None, "Override config directory path."] = None, 1130 branch: Annotated[str | None, "Initial branch name."] = None, 1131 ) -> "Dml": 1132 """Initialize a repository with an unborn attached HEAD.""" 1133 config = Config.init(project_home, remote_root=remote_root) 1134 dml = cls.from_config_vars( 1135 _python_config_vars_to_canonical( 1136 project_home=config.project_home, 1137 db_path=db_path, 1138 db_map_size_headroom=db_map_size_headroom, 1139 db_map_size_max=db_map_size_max, 1140 default_branch_name=default_branch_name, 1141 remote_root=remote_root, 1142 remote_prune_age_seconds=remote_prune_age_seconds, 1143 remote_fetch_workers=remote_fetch_workers, 1144 user=user, 1145 config_home=config_home, 1146 ) 1147 ) 1148 head = Head(config.project_home) 1149 branch = branch or dml._config.default.branch_name 1150 with head.lock(): 1151 try: 1152 head.get_head() 1153 except (DmlRepoError, FileNotFoundError): 1154 dml._db.init() 1155 head.init(None, branch) 1156 return dml
Initialize a repository with an unborn attached HEAD.
1158 @classmethod 1159 def clone( 1160 cls, 1161 revision: Annotated[Ref | str | None, "Optional branch, tag, or commit revision."] = None, 1162 /, 1163 *, 1164 project_home: Annotated[str, "Directory where the repository should be cloned."] = ".", 1165 db_path: Annotated[str | None, "Override path to the LMDB database."] = None, 1166 db_map_size_headroom: Annotated[int | None, "Extra LMDB map size headroom in bytes."] = None, 1167 db_map_size_max: Annotated[int | None, "Maximum LMDB map size in bytes."] = None, 1168 default_branch_name: Annotated[str | None, "Default branch name for attached HEAD operations."] = None, 1169 remote_root: Annotated[str | None, "Remote storage root URI."] = None, 1170 remote_prune_age_seconds: Annotated[int | None, "Remote GC prune age in seconds."] = None, 1171 remote_fetch_workers: Annotated[int | None, "Number of concurrent remote fetch workers."] = None, 1172 user: Annotated[str | None, "User name recorded in commits and runtime actions."] = None, 1173 config_home: Annotated[str | None, "Override config directory path."] = None, 1174 depth: Annotated[int | None, "Positive number of commit-history generations to fetch."] = None, 1175 ) -> "Dml": 1176 """Clone one revision from the configured remote root.""" 1177 _validate_history_options(depth) 1178 Path(project_home).mkdir(parents=True, exist_ok=True) 1179 config = Config.init(project_home, remote_root=remote_root) 1180 dml = cls.from_config_vars( 1181 _python_config_vars_to_canonical( 1182 project_home=config.project_home, 1183 db_path=db_path, 1184 db_map_size_headroom=db_map_size_headroom, 1185 db_map_size_max=db_map_size_max, 1186 default_branch_name=default_branch_name, 1187 remote_root=remote_root, 1188 remote_prune_age_seconds=remote_prune_age_seconds, 1189 remote_fetch_workers=remote_fetch_workers, 1190 user=user, 1191 config_home=config_home, 1192 ) 1193 ) 1194 _require_remote_root(dml) 1195 selected = revision or dml._config.default.branch_name 1196 exact = selected if isinstance(selected, Ref) else None 1197 if isinstance(selected, str) and re.match(r"^(?:commit:)?[0-9a-f]{64}$", selected): 1198 exact = Ref(selected if selected.startswith("commit:") else f"commit:{selected}") 1199 branch = ( 1200 selected 1201 if exact is None and isinstance(selected, str) and not selected.startswith(("@", "HEAD")) 1202 else None 1203 ) 1204 initial_branch = branch or dml._config.default.branch_name 1205 head = Head(config.project_home) 1206 with head.lock(): 1207 try: 1208 head.get_head() 1209 except (DmlRepoError, FileNotFoundError): 1210 dml._db.init() 1211 head.init(None, initial_branch) 1212 else: 1213 raise DmlRepoError(f"Cannot clone into an initialized repository: {dml._config.project_home}") 1214 if exact is not None: 1215 commit, available, omitted = _remote_ops(dml).materialize_project_commit_ref(exact, dml._db, depth=depth) 1216 with head.lock(): 1217 _publish_shallow_state(head, available, omitted) 1218 head.write_detached_head(commit) 1219 return dml 1220 dml.fetch(selected if isinstance(selected, str) else None, depth=depth) 1221 kind = "tag" if isinstance(selected, str) and selected.startswith("@") else "branch" 1222 name = selected[1:] if kind == "tag" and isinstance(selected, str) else selected 1223 assert isinstance(name, str) 1224 with head.lock(): 1225 commit = head.get_remote_tracking_ref(name, kind=kind) 1226 if kind == "tag": 1227 head.write_detached_head(commit) 1228 else: 1229 head.update_local_ref(name, commit, kind="branch") 1230 head.write_attached_head(name) 1231 head.set_upstream(name, name) 1232 return dml
Clone one revision from the configured remote root.
1234 def status(self) -> StatusPayload: 1235 """Return branch, commit, and open-runtime status for this repository.""" 1236 head = _head_ops(self) 1237 head_info = head.get_head() 1238 ahead = behind = None 1239 upstream = head.get_upstream(head_info["branch"]) if head_info["branch"] is not None else None 1240 if upstream is not None: 1241 try: 1242 upstream_ref = head.get_remote_tracking_ref(upstream["branch"]) 1243 except DmlRepoError: 1244 pass 1245 else: 1246 if head_info["commit"] is not None: 1247 try: 1248 ahead, behind = CommitOps().ahead_behind( 1249 head_info["commit"], 1250 upstream_ref, 1251 db=self._db, 1252 missing_commits=head.get_shallow_commits(), 1253 ) 1254 except ShallowHistoryError: 1255 pass 1256 with self._db.tx(readonly=True) as txn: 1257 num_indexes = 0 1258 for namespace in ("index", "frozenindex"): 1259 try: 1260 num_indexes += sum(1 for _ in txn.iter(namespace)) 1261 except DmlDbKeyNotFoundError: 1262 pass 1263 return { 1264 "mode": head_info["mode"], 1265 "branch": head_info["branch"], 1266 "commit": head_info["commit"], 1267 "branches": head.list_local_refs(kind="branch"), 1268 "upstream": upstream["branch"] if upstream is not None else None, 1269 "num_indexes": num_indexes, 1270 "ahead": ahead, 1271 "behind": behind, 1272 }
Return branch, commit, and open-runtime status for this repository.
1274 def log( 1275 self, 1276 revision: Annotated[Ref | str, "Revision to start the log from."] = "HEAD", 1277 limit: Annotated[int, "Maximum number of commits to return."] = 10, 1278 *, 1279 remote: Annotated[bool, "Resolve from fetched remote tracking refs."] = False, 1280 dep: Annotated[str | None, "Resolve from a fetched dependency."] = None, 1281 ) -> LogPayload: 1282 """Return commit history starting from one revision.""" 1283 commit_ref = resolve_rev(_head_ops(self), revision, db=self._db, remote=remote, dep=dep) 1284 commit_ref = _require_resolved_commit(commit_ref, revision) 1285 commits, truncated = CommitOps().log_with_truncation( 1286 commit_ref, 1287 limit=limit, 1288 db=self._db, 1289 missing_commits=_head_ops(self).get_shallow_commits(), 1290 ) 1291 return {"commits": commits, "truncated": truncated}
Return commit history starting from one revision.
1293 def show( 1294 self, 1295 revision: Annotated[Ref | str, "Revision to describe."] = "HEAD", 1296 *, 1297 remote: bool = False, 1298 dep: str | None = None, 1299 ) -> CommitFullDescription: 1300 """Return a full commit description for one revision.""" 1301 commit_ref = resolve_rev(_head_ops(self), revision, db=self._db, remote=remote, dep=dep) 1302 commit_ref = _require_resolved_commit(commit_ref, revision) 1303 return CommitOps().show( 1304 commit_ref, 1305 db=self._db, 1306 missing_commits=_head_ops(self).get_shallow_commits(), 1307 )
Return a full commit description for one revision.
1309 def diff( 1310 self, 1311 revision: Annotated[Ref | str, "Revision to diff."] = "HEAD", 1312 relative_to: Annotated[Ref | str | None, "Optional base revision. Defaults to the commit parent."] = None, 1313 *, 1314 remote: Annotated[bool, "Resolve the primary revision from remote tracking."] = False, 1315 dep: Annotated[str | None, "Resolve the primary revision from a dependency."] = None, 1316 ) -> CommitDiffPayload: 1317 """Return DAG-level changes for one revision.""" 1318 head = _head_ops(self) 1319 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote, dep=dep) 1320 commit_ref = _require_resolved_commit(commit_ref, revision) 1321 if relative_to is None: 1322 return CommitOps().diff( 1323 commit_ref, 1324 db=self._db, 1325 missing_commits=head.get_shallow_commits(), 1326 ) 1327 rel_to_commit = resolve_rev(head, relative_to, db=self._db) 1328 rel_to_commit = _require_resolved_commit(rel_to_commit, relative_to) 1329 return CommitOps().diff( 1330 commit_ref, 1331 rel_to_commit, 1332 db=self._db, 1333 missing_commits=head.get_shallow_commits(), 1334 )
Return DAG-level changes for one revision.
1336 def rev_parse( 1337 self, 1338 revision: Annotated[str, "Revision expression to resolve."], 1339 *, 1340 remote: bool = False, 1341 dep: str | None = None, 1342 ) -> RevisionPayload: 1343 """Resolve a revision expression into a commit and ref metadata.""" 1344 head = _head_ops(self) 1345 branch = tag = None 1346 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote, dep=dep) 1347 if revision.startswith("HEAD"): 1348 kind = "head" 1349 elif re.match(r"^(?:commit:)?[0-9a-f]{64}$", revision): 1350 kind = "commit" 1351 else: 1352 kind = "ref" 1353 if kind == "ref": 1354 branch, tag = (None, revision[1:]) if revision.startswith("@") else (revision, None) 1355 return { 1356 "input": revision, 1357 "uri": None, 1358 "kind": kind, 1359 "commit": commit_ref, 1360 "branch": branch, 1361 "tag": tag, 1362 }
Resolve a revision expression into a commit and ref metadata.
1364 def revert( 1365 self, 1366 revision: Annotated[Ref | str, "Revision whose changes should be reverted."], 1367 message: Annotated[str | None, "Optional commit message for the revert commit."] = None, 1368 *, 1369 remote: bool = False, 1370 ) -> StatusPayload: 1371 """Revert the changes introduced by one revision.""" 1372 head = _head_ops(self) 1373 with head.lock(): 1374 head_info = head.get_head() 1375 if head_info["branch"] is None: 1376 raise DmlRepoError("Cannot revert when HEAD is detached") 1377 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote) 1378 commit_ref = _require_resolved_commit(commit_ref, revision) 1379 new_commit = CommitOps().revert( 1380 commit_ref, 1381 _require_resolved_commit(head_info["commit"], "HEAD"), 1382 user=self._config.user, 1383 message=message, 1384 db=self._db, 1385 missing_commits=head.get_shallow_commits(), 1386 ) 1387 head.update_local_ref(head_info["branch"], new_commit, kind="branch") 1388 return self.status()
Revert the changes introduced by one revision.
1390 def checkout( 1391 self, revision: Annotated[Ref | str, "Revision to check out."], *, remote: bool = False 1392 ) -> StatusPayload: 1393 """Check out a different revision. 1394 1395 If the revision resolves to a local branch, HEAD stays attached to that branch. 1396 Other revisions detach HEAD at the resolved commit. 1397 """ 1398 head = _head_ops(self) 1399 with head.lock(): 1400 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote) 1401 if ( 1402 not remote 1403 and isinstance(revision, str) 1404 and not revision.startswith(("@", "HEAD")) 1405 and not re.match(r"^(?:commit:)?[0-9a-f]{64}$", revision) 1406 ): 1407 head.write_attached_head(revision) 1408 else: 1409 head.write_detached_head(_require_resolved_commit(commit_ref, revision)) 1410 return self.status()
Check out a different revision.
If the revision resolves to a local branch, HEAD stays attached to that branch. Other revisions detach HEAD at the resolved commit.
1412 def merge( 1413 self, 1414 revision: Annotated[Ref | str, "Revision to merge into the current branch."], 1415 ff_only: Annotated[bool, "Whether to only allow fast-forward merges."] = True, 1416 *, 1417 remote: bool = False, 1418 ) -> StatusPayload: 1419 """Merge a revision into the current HEAD.""" 1420 head = _head_ops(self) 1421 with head.lock(): 1422 head_info = head.get_head() 1423 if head_info["branch"] is None: 1424 raise DmlRepoError("Cannot merge when HEAD is detached") 1425 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote) 1426 commit_ref = _require_resolved_commit(commit_ref, revision) 1427 new_commit = CommitOps().merge( 1428 head_info["commit"], 1429 commit_ref, 1430 user=self._config.user, 1431 ff_only=ff_only, 1432 db=self._db, 1433 missing_commits=head.get_shallow_commits(), 1434 ) 1435 head.update_local_ref(head_info["branch"], new_commit) 1436 return self.status()
Merge a revision into the current HEAD.
1438 def rebase( 1439 self, revision: Annotated[Ref | str, "Revision to rebase the current branch onto."], *, remote: bool = False 1440 ) -> StatusPayload: 1441 """Rebase the current HEAD onto a different revision.""" 1442 head = _head_ops(self) 1443 with head.lock(): 1444 head_info = head.get_head() 1445 if head_info["branch"] is None: 1446 raise DmlRepoError("Cannot rebase when HEAD is detached") 1447 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote) 1448 new_commit = CommitOps().rebase( 1449 _require_resolved_commit(head_info["commit"], "HEAD"), 1450 _require_resolved_commit(commit_ref, revision), 1451 user=self._config.user, 1452 db=self._db, 1453 missing_commits=head.get_shallow_commits(), 1454 ) 1455 head.update_local_ref(head_info["branch"], new_commit) 1456 return self.status()
Rebase the current HEAD onto a different revision.
1458 def fetch( 1459 self, 1460 revision: Annotated[str | None, "Branch or @tag to fetch."] = None, 1461 /, 1462 *, 1463 dep: Annotated[str | None, "Named dependency endpoint to fetch from."] = None, 1464 depth: Annotated[int | None, "Positive number of commit-history generations to fetch."] = None, 1465 unshallow: Annotated[bool, "Fetch all history through existing shallow boundaries."] = False, 1466 ) -> None: 1467 """Fetch one branch or tag from remote.root or a named dependency.""" 1468 _validate_history_options(depth, unshallow) 1469 selector = revision or self._config.default.branch_name 1470 kind = "tag" if selector.startswith("@") else "branch" 1471 name = selector[1:] if kind == "tag" else selector 1472 head = _head_ops(self) 1473 if dep is None: 1474 remote_ops = _remote_ops(self) 1475 else: 1476 config = head.get_dependency_config(dep) 1477 remote_ops = Remote( 1478 config["root"], 1479 n_workers=self._config.remote.fetch_workers, 1480 client=_require_s3_client(self), 1481 prune_age_seconds=self._config.remote.prune_age_seconds, 1482 ) 1483 materialized = remote_ops.get_project_commit_ref( 1484 kind, 1485 name, 1486 db=self._db, 1487 depth=depth, 1488 unshallow=unshallow, 1489 ) 1490 if materialized is None: 1491 raise DmlRepoError(f"Remote {kind} ref not found: {selector}") 1492 commit, available, omitted = materialized 1493 with head.lock(): 1494 _publish_shallow_state(head, available, omitted) 1495 if dep is None: 1496 head.update_remote_tracking_ref(name, commit, kind=kind) 1497 else: 1498 head.update_dependency_ref(dep, name, commit, kind=kind)
Fetch one branch or tag from remote.root or a named dependency.
1500 def pull( 1501 self, 1502 ff_only: Annotated[bool, "Whether to only allow fast-forward merges."] = True, 1503 *, 1504 depth: Annotated[int | None, "Positive number of commit-history generations to fetch."] = None, 1505 ) -> StatusPayload: 1506 """Fetch and merge the current branch's configured upstream.""" 1507 _validate_history_options(depth) 1508 head = _head_ops(self) 1509 head_info = head.get_head() 1510 if head_info["branch"] is None: 1511 raise DmlRepoError("Cannot pull when HEAD is detached") 1512 upstream = head.get_upstream(head_info["branch"]) 1513 if upstream is None: 1514 raise DmlRepoError(f"Cannot pull untracked branch: {head_info['branch']}") 1515 self.fetch(upstream["branch"], depth=depth) 1516 self.merge(upstream["branch"], ff_only=ff_only, remote=True) 1517 return self.status()
Fetch and merge the current branch's configured upstream.
1519 def push( 1520 self, 1521 *, 1522 revision: Annotated[Ref | str, "Local revision to publish; defaults to HEAD."] = "HEAD", 1523 force: Annotated[bool, "Overwrite a remote branch or tag without publication checks."] = False, 1524 ) -> None: 1525 """Publish a local revision as a tag or branch on the configured remote.""" 1526 head = _head_ops(self) 1527 commit_ref = _require_resolved_commit(resolve_rev(head, revision, db=self._db), revision) 1528 if isinstance(revision, str) and revision.startswith("@"): 1529 _remote_ops(self).put_ref( 1530 commit_ref, 1531 kind="tag", 1532 name=revision[1:], 1533 db=self._db, 1534 force=force, 1535 missing_commits=head.get_shallow_commits(), 1536 ) 1537 return 1538 named_branch = ( 1539 isinstance(revision, str) 1540 and not revision.startswith("HEAD") 1541 and re.fullmatch(r"(?:commit:)?[0-9a-f]{64}", revision) is None 1542 ) 1543 branch = revision if named_branch and isinstance(revision, str) else head.get_head()["branch"] 1544 if branch is None: 1545 raise DmlRepoError("Cannot push an unnamed revision when HEAD is detached") 1546 upstream = head.get_upstream(branch) 1547 if upstream is None: 1548 upstream = {"branch": branch} 1549 set_upstream = True 1550 else: 1551 set_upstream = False 1552 remote = _remote_ops(self) 1553 remote.put_ref( 1554 commit_ref, 1555 kind="branch", 1556 name=upstream["branch"], 1557 db=self._db, 1558 force=force, 1559 missing_commits=head.get_shallow_commits(), 1560 ) 1561 if set_upstream: 1562 with head.lock(): 1563 head.set_upstream(branch, upstream["branch"])
Publish a local revision as a tag or branch on the configured remote.
1565 def gc( 1566 self, 1567 *, 1568 remote: Annotated[bool, "Garbage-collect configured remote state instead of local objects."] = False, 1569 unrecognized: Annotated[ 1570 Literal["raise", "ignore"], 1571 "Remote only: fail on (raise) or log and skip (ignore) unrecognized objects in immutable storage.", 1572 ] = "raise", 1573 ) -> LocalGCSummary | RemoteGCSummary: 1574 """Garbage-collect unreachable local or configured remote state.""" 1575 if not remote: 1576 return _local_gc(self) 1577 return cast(RemoteGCSummary, _remote_ops(self).gc(unrecognized))
Garbage-collect unreachable local or configured remote state.
Expose branch inspection and lifecycle commands.
Expose import-only dependency lifecycle commands.
Expose remote execution cache inspection and control commands.
Expose tag inspection and lifecycle commands.
Expose configuration commands.
Expose runtime mutation and execution-state inspection commands.
Expose committed DAG inspection commands.
395@_register_dml_obj 396class Error(DmlBase, Exception): 397 """Error information with stack traces. 398 399 Represents a captured error from a computation, storing error details 400 and stack trace information for debugging. 401 402 Attributes 403 ---------- 404 message : str 405 The error message. 406 origin : str 407 The origin/source of the error (e.g., 'python', 'adapter'). 408 type : str 409 The error type name. 410 stack : list[dict] 411 Stack trace frames as dictionaries. 412 """ 413 414 message: str 415 origin: str 416 type: str 417 stack: list[dict] = field(default_factory=list) 418 419 def __post_init__(self): 420 """Initialize Exception base with message and run base initialization.""" 421 Exception.__init__(self, self.message) 422 423 def _validate(self) -> None: 424 if not isinstance(self.message, str): 425 raise TypeError(f"{self.__class__.__name__}.message must be a string") 426 if not isinstance(self.origin, str): 427 raise TypeError(f"{self.__class__.__name__}.origin must be a string") 428 if not isinstance(self.type, str): 429 raise TypeError(f"{self.__class__.__name__}.type must be a string") 430 if not isinstance(self.stack, list): 431 raise TypeError(f"{self.__class__.__name__}.stack must be a list of frame dicts") 432 for frame in self.stack: 433 if not isinstance(frame, dict): 434 raise TypeError(f"{self.__class__.__name__}.stack frame must be a dict") 435 436 @classmethod 437 def from_ex(cls, exc) -> "Error": 438 """Create Error from Python exception. 439 440 Parameters 441 ---------- 442 exc : Exception 443 Python exception to convert. 444 445 Returns 446 ------- 447 Error 448 Error object with extracted stack trace. 449 """ 450 if isinstance(exc, Error): 451 return Error(message=exc.message, origin=exc.origin, type=exc.type, stack=list(exc.stack)) 452 tb = traceback.extract_tb(exc.__traceback__) 453 stack = [ 454 { 455 "filename": frame.filename, 456 "lineno": frame.lineno, 457 "name": frame.name, 458 "line": frame.line, 459 } 460 for frame in tb 461 ] 462 return cls( 463 message=str(exc), 464 origin="python", 465 type=type(exc).__name__.lower(), 466 stack=stack, 467 )
Error information with stack traces.
Represents a captured error from a computation, storing error details and stack trace information for debugging.
436 @classmethod 437 def from_ex(cls, exc) -> "Error": 438 """Create Error from Python exception. 439 440 Parameters 441 ---------- 442 exc : Exception 443 Python exception to convert. 444 445 Returns 446 ------- 447 Error 448 Error object with extracted stack trace. 449 """ 450 if isinstance(exc, Error): 451 return Error(message=exc.message, origin=exc.origin, type=exc.type, stack=list(exc.stack)) 452 tb = traceback.extract_tb(exc.__traceback__) 453 stack = [ 454 { 455 "filename": frame.filename, 456 "lineno": frame.lineno, 457 "name": frame.name, 458 "line": frame.line, 459 } 460 for frame in tb 461 ] 462 return cls( 463 message=str(exc), 464 origin="python", 465 type=type(exc).__name__.lower(), 466 stack=stack, 467 )
Create Error from Python exception.
62class ExecutionSemanticState(TypedDict): 63 lifecycle: EXECUTION_LIFECYCLES 64 result_ref: str | None 65 result_source: Literal["runtime", "adapter-error"] | None 66 spawned_execution_ids: list[str] 67 child_execution_ids: list[str] 68 cancelation: ControlRecord | None 69 invalidation: ControlRecord | None 70 updated_at: int
678@dataclass(frozen=True) 679class Node: # noqa: F811 680 """ 681 Representation of a node in a DaggerML DAG. 682 683 Parameters 684 ---------- 685 dag : Dag 686 Parent DAG 687 ref : Ref 688 Node reference 689 """ 690 691 dag: Dag 692 ref: Ref 693 _info: dict = field(default_factory=dict) 694 695 def __repr__(self): 696 ref_id = self.ref if isinstance(self.ref, Error) else self.ref.to 697 return f"{self.__class__.__name__}({ref_id})" 698 699 def __hash__(self): 700 return hash(self.ref) 701 702 def __eq__(self, other): 703 if not isinstance(other, Node): 704 return NotImplemented 705 return self.ref == other.ref 706 707 def context(self, *, root: bool = True) -> Dag: 708 """ 709 Resolve the provenance context (DAG) for this node. 710 711 This follows import/function provenance while treating builtin 712 collection-construction and selection DAGs as transparent. 713 714 Parameters 715 ---------- 716 root : bool, default=True 717 If False, return the nearest sub-DAG in which this value exists as a 718 proper node across a non-builtin import/function boundary. If True, 719 continue recursively until provenance no longer crosses a 720 non-builtin import/function boundary and return that first rooted 721 context. 722 723 Returns 724 ------- 725 Dag 726 The nearest or rooted provenance DAG for this node. 727 728 Examples 729 -------- 730 >>> source = new(dml=dml, name="source", message="source") 731 >>> answer = source.put(42, name="answer") 732 >>> payload = source.put({"answer": answer}, name="payload") 733 >>> source.commit(payload) 734 >>> consumer = new(dml=dml, name="consumer", message="consumer") 735 >>> imported = consumer.require("source", "payload", name="payload") 736 >>> consumer.commit(imported) 737 >>> loaded = load("consumer", dml=dml) 738 >>> loaded.result["answer"].context() == source 739 True 740 """ 741 return _resolve_context(self, (), root=root) 742 743 @property 744 def type(self): 745 """Get the data type of the node.""" 746 return self._info["data_type"] 747 748 @overload 749 def value(self: "ScalarNode") -> Scalar: ... 750 @overload 751 def value(self: "ListNode") -> list: ... 752 @overload 753 def value(self: "DictNode") -> dict: ... 754 @overload 755 def value(self: "RunnableNode") -> Runnable: ... 756 @overload 757 def value(self: "Node") -> Any: ... 758 def value(self): 759 """ 760 Get the concrete value of this node. 761 762 Returns 763 ------- 764 Any 765 The actual value represented by this node 766 """ 767 value = self.dag.dml.dag.get_node(self.ref, recursive=True) 768 if isinstance(value, Error): 769 raise NodeError(value, node_ref=self.ref, dag=self.dag) 770 return value 771 772 def __call__(self, *args, name=None, sleep=None, timeout=-1, **kw) -> "Node": 773 raise TypeError(f"Node of type '{self.type}' is not callable")
Representation of a node in a DaggerML DAG.
707 def context(self, *, root: bool = True) -> Dag: 708 """ 709 Resolve the provenance context (DAG) for this node. 710 711 This follows import/function provenance while treating builtin 712 collection-construction and selection DAGs as transparent. 713 714 Parameters 715 ---------- 716 root : bool, default=True 717 If False, return the nearest sub-DAG in which this value exists as a 718 proper node across a non-builtin import/function boundary. If True, 719 continue recursively until provenance no longer crosses a 720 non-builtin import/function boundary and return that first rooted 721 context. 722 723 Returns 724 ------- 725 Dag 726 The nearest or rooted provenance DAG for this node. 727 728 Examples 729 -------- 730 >>> source = new(dml=dml, name="source", message="source") 731 >>> answer = source.put(42, name="answer") 732 >>> payload = source.put({"answer": answer}, name="payload") 733 >>> source.commit(payload) 734 >>> consumer = new(dml=dml, name="consumer", message="consumer") 735 >>> imported = consumer.require("source", "payload", name="payload") 736 >>> consumer.commit(imported) 737 >>> loaded = load("consumer", dml=dml) 738 >>> loaded.result["answer"].context() == source 739 True 740 """ 741 return _resolve_context(self, (), root=root)
Resolve the provenance context (DAG) for this node.
This follows import/function provenance while treating builtin collection-construction and selection DAGs as transparent.
>>> source = new(dml=dml, name="source", message="source")
>>> answer = source.put(42, name="answer")
>>> payload = source.put({"answer": answer}, name="payload")
>>> source.commit(payload)
>>> consumer = new(dml=dml, name="consumer", message="consumer")
>>> imported = consumer.require("source", "payload", name="payload")
>>> consumer.commit(imported)
>>> loaded = load("consumer", dml=dml)
>>> loaded.result["answer"].context() == source
True
758 def value(self): 759 """ 760 Get the concrete value of this node. 761 762 Returns 763 ------- 764 Any 765 The actual value represented by this node 766 """ 767 value = self.dag.dml.dag.get_node(self.ref, recursive=True) 768 if isinstance(value, Error): 769 raise NodeError(value, node_ref=self.ref, dag=self.dag) 770 return value
Get the concrete value of this node.
Reference to another node.
Ref distinguishes stored references from plain strings, which is needed
so serialization can round-trip graph edges instead of raw text.
Initialize a reference wrapper.
namespace:id form.to is not a string.Return the namespace portion of the reference.
This uses the database C parser so Python and C agree on ref structure.
Return the identifier portion of the reference.
This complements ns() by exposing the ID while keeping the split
logic centralized in the database layer.
319@dataclass 320class Runnable: 321 target: Uri 322 sub: Optional["Runnable"] = None 323 kwargs: dict[str, Any] = field(default_factory=dict) 324 adapter: str = "" 325 326 def innermost(self) -> "Runnable": 327 """Get the innermost Runnable in the chain.""" 328 current = self 329 while current.sub is not None: 330 current = current.sub 331 return current
Return the active default Dml runtime.
Set the process-default Dml runtime.
Temporarily override the default Dml runtime for the active context.
Clear the process-default Dml runtime.
74def new( 75 name="", 76 message="", 77 cache_key: str | None = None, 78 execution_id: str | None = None, 79 tags: list[str] | None = None, 80 dml: Dml | None = None, 81) -> "Dag": 82 """Create a new DAG using the active or provided Dml runtime.""" 83 runtime = dml or get_default_dml() 84 execution = Ref(f"index:{execution_id}") if execution_id is not None else None 85 create_kwargs = {"cache_key": cache_key, "execution": execution} 86 if tags is not None: 87 create_kwargs["tags"] = tags 88 index_id = runtime.runtime.create(**create_kwargs) 89 return Dag(dml=runtime, token=index_id, name=name, message=message)
Create a new DAG using the active or provided Dml runtime.
92def load( 93 name: str, 94 dml: Dml | None = None, 95 *, 96 revision: Ref | str = "HEAD", 97 remote: bool = False, 98 dep: str | None = None, 99) -> "Dag": 100 """Load a DAG using the active default Dml runtime.""" 101 dml = dml or get_default_dml() 102 dag_ref = dml.show(revision, remote=remote, dep=dep)["dags"].get(name) 103 if dag_ref is None: 104 raise DmlRepoError(f"DAG not found: {name}") 105 return Dag(dml=dml, ref=dag_ref, name=name)
Load a DAG using the active default Dml runtime.
108def resume( 109 frozen: Ref, 110 *, 111 name: str, 112 message: str, 113 dml: Dml | None = None, 114) -> "Dag": 115 """Resume a frozen DAG runtime with explicitly supplied commit metadata.""" 116 runtime = dml or get_default_dml() 117 index_id = runtime.runtime.unfreeze(frozen) 118 return Dag(dml=runtime, token=index_id, name=name, message=message)
Resume a frozen DAG runtime with explicitly supplied commit metadata.
128def status() -> dict[str, object]: 129 """Return status for the active default Dml runtime.""" 130 dml, source = _resolve_default_dml(create=True) 131 return { 132 "default": { 133 "source": source, 134 "has_scoped_override": _SCOPED_DEFAULT_DML.get() is not _NO_DEFAULT_DML, 135 "has_process_default": _PROCESS_DEFAULT_DML is not None, 136 }, 137 "status": dml.status(), 138 }
Return status for the active default Dml runtime.
Create a temporary Dml runtime with an unborn attached HEAD.