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.
1074class Dml: 1075 def _init_from_config_vars(self, explicit_config: Mapping[str, object]) -> None: 1076 self._explicit_config = dict(explicit_config) 1077 self._config = Config.resolve(explicit=self._explicit_config) 1078 dflt = self._config.default 1079 self._db = DmlDB(self._config.db_path, dflt.db_map_size_headroom, dflt.db_map_size_max) 1080 self._s3_client = None 1081 1082 def __init__( 1083 self, 1084 project_home: Annotated[str | None, "Project root containing the .dml repository."] = None, 1085 *, 1086 db_path: Annotated[str | None, "Override path to the LMDB database."] = None, 1087 db_map_size_headroom: Annotated[int | None, "Extra LMDB map size headroom in bytes."] = None, 1088 db_map_size_max: Annotated[int | None, "Maximum LMDB map size in bytes."] = None, 1089 default_branch_name: Annotated[str | None, "Default branch name for attached HEAD operations."] = None, 1090 remote_root: Annotated[str | None, "Remote storage root URI."] = None, 1091 remote_prune_age_seconds: Annotated[int | None, "Remote GC prune age in seconds."] = None, 1092 remote_fetch_workers: Annotated[int | None, "Number of concurrent remote fetch workers."] = None, 1093 user: Annotated[str | None, "User name recorded in commits and runtime actions."] = None, 1094 config_home: Annotated[str | None, "Override config directory path."] = None, 1095 ): 1096 """Create a DaggerML session bound to one repository and config context.""" 1097 self._init_from_config_vars( 1098 _python_config_vars_to_canonical( 1099 project_home=project_home, 1100 db_path=db_path, 1101 db_map_size_headroom=db_map_size_headroom, 1102 db_map_size_max=db_map_size_max, 1103 default_branch_name=default_branch_name, 1104 remote_root=remote_root, 1105 remote_prune_age_seconds=remote_prune_age_seconds, 1106 remote_fetch_workers=remote_fetch_workers, 1107 user=user, 1108 config_home=config_home, 1109 ) 1110 ) 1111 1112 @classmethod 1113 def from_config_vars( 1114 cls, 1115 config_vars: Annotated[dict[str, object], "Flattened canonical config-var mapping."] | None = None, 1116 ) -> "Dml": 1117 """Create a DaggerML session from flattened canonical config vars.""" 1118 dml = cls.__new__(cls) 1119 dml._init_from_config_vars(config_vars or {}) 1120 return dml 1121 1122 @classmethod 1123 def init( 1124 cls, 1125 project_home: Annotated[str, "Directory where the repository should be initialized."] = ".", 1126 *, 1127 db_path: Annotated[str | None, "Override path to the LMDB database."] = None, 1128 db_map_size_headroom: Annotated[int | None, "Extra LMDB map size headroom in bytes."] = None, 1129 db_map_size_max: Annotated[int | None, "Maximum LMDB map size in bytes."] = None, 1130 default_branch_name: Annotated[str | None, "Default branch name for attached HEAD operations."] = None, 1131 remote_root: Annotated[str | None, "Remote storage root URI."] = None, 1132 remote_prune_age_seconds: Annotated[int | None, "Remote GC prune age in seconds."] = None, 1133 remote_fetch_workers: Annotated[int | None, "Number of concurrent remote fetch workers."] = None, 1134 user: Annotated[str | None, "User name recorded in commits and runtime actions."] = None, 1135 config_home: Annotated[str | None, "Override config directory path."] = None, 1136 branch: Annotated[str | None, "Initial branch name."] = None, 1137 ) -> "Dml": 1138 """Initialize a repository with an unborn attached HEAD.""" 1139 config = Config.init(project_home, remote_root=remote_root) 1140 dml = cls.from_config_vars( 1141 _python_config_vars_to_canonical( 1142 project_home=config.project_home, 1143 db_path=db_path, 1144 db_map_size_headroom=db_map_size_headroom, 1145 db_map_size_max=db_map_size_max, 1146 default_branch_name=default_branch_name, 1147 remote_root=remote_root, 1148 remote_prune_age_seconds=remote_prune_age_seconds, 1149 remote_fetch_workers=remote_fetch_workers, 1150 user=user, 1151 config_home=config_home, 1152 ) 1153 ) 1154 head = Head(config.project_home) 1155 branch = branch or dml._config.default.branch_name 1156 with head.lock(): 1157 try: 1158 head.get_head() 1159 except (DmlRepoError, FileNotFoundError): 1160 dml._db.init() 1161 head.init(None, branch) 1162 return dml 1163 1164 @classmethod 1165 def clone( 1166 cls, 1167 revision: Annotated[Ref | str | None, "Optional branch, tag, or commit revision."] = None, 1168 /, 1169 *, 1170 project_home: Annotated[str, "Directory where the repository should be cloned."] = ".", 1171 db_path: Annotated[str | None, "Override path to the LMDB database."] = None, 1172 db_map_size_headroom: Annotated[int | None, "Extra LMDB map size headroom in bytes."] = None, 1173 db_map_size_max: Annotated[int | None, "Maximum LMDB map size in bytes."] = None, 1174 default_branch_name: Annotated[str | None, "Default branch name for attached HEAD operations."] = None, 1175 remote_root: Annotated[str | None, "Remote storage root URI."] = None, 1176 remote_prune_age_seconds: Annotated[int | None, "Remote GC prune age in seconds."] = None, 1177 remote_fetch_workers: Annotated[int | None, "Number of concurrent remote fetch workers."] = None, 1178 user: Annotated[str | None, "User name recorded in commits and runtime actions."] = None, 1179 config_home: Annotated[str | None, "Override config directory path."] = None, 1180 depth: Annotated[int | None, "Positive number of commit-history generations to fetch."] = None, 1181 ) -> "Dml": 1182 """Clone one revision from the configured remote root.""" 1183 _validate_history_options(depth) 1184 Path(project_home).mkdir(parents=True, exist_ok=True) 1185 config = Config.init(project_home, remote_root=remote_root) 1186 dml = cls.from_config_vars( 1187 _python_config_vars_to_canonical( 1188 project_home=config.project_home, 1189 db_path=db_path, 1190 db_map_size_headroom=db_map_size_headroom, 1191 db_map_size_max=db_map_size_max, 1192 default_branch_name=default_branch_name, 1193 remote_root=remote_root, 1194 remote_prune_age_seconds=remote_prune_age_seconds, 1195 remote_fetch_workers=remote_fetch_workers, 1196 user=user, 1197 config_home=config_home, 1198 ) 1199 ) 1200 _require_remote_root(dml) 1201 selected = revision or dml._config.default.branch_name 1202 exact = selected if isinstance(selected, Ref) else None 1203 if isinstance(selected, str) and re.match(r"^(?:commit:)?[0-9a-f]{64}$", selected): 1204 exact = Ref(selected if selected.startswith("commit:") else f"commit:{selected}") 1205 branch = ( 1206 selected 1207 if exact is None and isinstance(selected, str) and not selected.startswith(("@", "HEAD")) 1208 else None 1209 ) 1210 initial_branch = branch or dml._config.default.branch_name 1211 head = Head(config.project_home) 1212 with head.lock(): 1213 try: 1214 head.get_head() 1215 except (DmlRepoError, FileNotFoundError): 1216 dml._db.init() 1217 head.init(None, initial_branch) 1218 else: 1219 raise DmlRepoError(f"Cannot clone into an initialized repository: {dml._config.project_home}") 1220 if exact is not None: 1221 commit, available, omitted = _remote_ops(dml).materialize_project_commit_ref(exact, dml._db, depth=depth) 1222 with head.lock(): 1223 _publish_shallow_state(head, available, omitted) 1224 head.write_detached_head(commit) 1225 return dml 1226 dml.fetch(selected if isinstance(selected, str) else None, depth=depth) 1227 kind = "tag" if isinstance(selected, str) and selected.startswith("@") else "branch" 1228 name = selected[1:] if kind == "tag" and isinstance(selected, str) else selected 1229 assert isinstance(name, str) 1230 with head.lock(): 1231 commit = head.get_remote_tracking_ref(name, kind=kind) 1232 if kind == "tag": 1233 head.write_detached_head(commit) 1234 else: 1235 head.update_local_ref(name, commit, kind="branch") 1236 head.write_attached_head(name) 1237 head.set_upstream(name, name) 1238 return dml 1239 1240 def status(self) -> StatusPayload: 1241 """Return branch, commit, and open-runtime status for this repository.""" 1242 head = _head_ops(self) 1243 head_info = head.get_head() 1244 ahead = behind = None 1245 upstream = head.get_upstream(head_info["branch"]) if head_info["branch"] is not None else None 1246 if upstream is not None: 1247 try: 1248 upstream_ref = head.get_remote_tracking_ref(upstream["branch"]) 1249 except DmlRepoError: 1250 pass 1251 else: 1252 if head_info["commit"] is not None: 1253 try: 1254 ahead, behind = CommitOps().ahead_behind( 1255 head_info["commit"], 1256 upstream_ref, 1257 db=self._db, 1258 missing_commits=head.get_shallow_commits(), 1259 ) 1260 except ShallowHistoryError: 1261 pass 1262 with self._db.tx(readonly=True) as txn: 1263 num_indexes = 0 1264 for namespace in ("index", "frozenindex"): 1265 try: 1266 num_indexes += sum(1 for _ in txn.iter(namespace)) 1267 except DmlDbKeyNotFoundError: 1268 pass 1269 return { 1270 "mode": head_info["mode"], 1271 "branch": head_info["branch"], 1272 "commit": head_info["commit"], 1273 "branches": head.list_local_refs(kind="branch"), 1274 "upstream": upstream["branch"] if upstream is not None else None, 1275 "num_indexes": num_indexes, 1276 "ahead": ahead, 1277 "behind": behind, 1278 } 1279 1280 def log( 1281 self, 1282 revision: Annotated[Ref | str, "Revision to start the log from."] = "HEAD", 1283 limit: Annotated[int, "Maximum number of commits to return."] = 10, 1284 *, 1285 remote: Annotated[bool, "Resolve from fetched remote tracking refs."] = False, 1286 dep: Annotated[str | None, "Resolve from a fetched dependency."] = None, 1287 ) -> LogPayload: 1288 """Return commit history starting from one revision.""" 1289 commit_ref = resolve_rev(_head_ops(self), revision, db=self._db, remote=remote, dep=dep) 1290 commit_ref = _require_resolved_commit(commit_ref, revision) 1291 commits, truncated = CommitOps().log_with_truncation( 1292 commit_ref, 1293 limit=limit, 1294 db=self._db, 1295 missing_commits=_head_ops(self).get_shallow_commits(), 1296 ) 1297 return {"commits": commits, "truncated": truncated} 1298 1299 def show( 1300 self, 1301 revision: Annotated[Ref | str, "Revision to describe."] = "HEAD", 1302 *, 1303 remote: bool = False, 1304 dep: str | None = None, 1305 ) -> CommitFullDescription: 1306 """Return a full commit description for one revision.""" 1307 commit_ref = resolve_rev(_head_ops(self), revision, db=self._db, remote=remote, dep=dep) 1308 commit_ref = _require_resolved_commit(commit_ref, revision) 1309 return CommitOps().show( 1310 commit_ref, 1311 db=self._db, 1312 missing_commits=_head_ops(self).get_shallow_commits(), 1313 ) 1314 1315 def diff( 1316 self, 1317 revision: Annotated[Ref | str, "Revision to diff."] = "HEAD", 1318 relative_to: Annotated[Ref | str | None, "Optional base revision. Defaults to the commit parent."] = None, 1319 *, 1320 remote: Annotated[bool, "Resolve the primary revision from remote tracking."] = False, 1321 dep: Annotated[str | None, "Resolve the primary revision from a dependency."] = None, 1322 ) -> CommitDiffPayload: 1323 """Return DAG-level changes for one revision.""" 1324 head = _head_ops(self) 1325 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote, dep=dep) 1326 commit_ref = _require_resolved_commit(commit_ref, revision) 1327 if relative_to is None: 1328 return CommitOps().diff( 1329 commit_ref, 1330 db=self._db, 1331 missing_commits=head.get_shallow_commits(), 1332 ) 1333 rel_to_commit = resolve_rev(head, relative_to, db=self._db) 1334 rel_to_commit = _require_resolved_commit(rel_to_commit, relative_to) 1335 return CommitOps().diff( 1336 commit_ref, 1337 rel_to_commit, 1338 db=self._db, 1339 missing_commits=head.get_shallow_commits(), 1340 ) 1341 1342 def rev_parse( 1343 self, 1344 revision: Annotated[str, "Revision expression to resolve."], 1345 *, 1346 remote: bool = False, 1347 dep: str | None = None, 1348 ) -> RevisionPayload: 1349 """Resolve a revision expression into a commit and ref metadata.""" 1350 head = _head_ops(self) 1351 branch = tag = None 1352 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote, dep=dep) 1353 if revision.startswith("HEAD"): 1354 kind = "head" 1355 elif re.match(r"^(?:commit:)?[0-9a-f]{64}$", revision): 1356 kind = "commit" 1357 else: 1358 kind = "ref" 1359 if kind == "ref": 1360 branch, tag = (None, revision[1:]) if revision.startswith("@") else (revision, None) 1361 return { 1362 "input": revision, 1363 "uri": None, 1364 "kind": kind, 1365 "commit": commit_ref, 1366 "branch": branch, 1367 "tag": tag, 1368 } 1369 1370 def revert( 1371 self, 1372 revision: Annotated[Ref | str, "Revision whose changes should be reverted."], 1373 message: Annotated[str | None, "Optional commit message for the revert commit."] = None, 1374 *, 1375 remote: bool = False, 1376 ) -> StatusPayload: 1377 """Revert the changes introduced by one revision.""" 1378 head = _head_ops(self) 1379 with head.lock(): 1380 head_info = head.get_head() 1381 if head_info["branch"] is None: 1382 raise DmlRepoError("Cannot revert when HEAD is detached") 1383 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote) 1384 commit_ref = _require_resolved_commit(commit_ref, revision) 1385 new_commit = CommitOps().revert( 1386 commit_ref, 1387 _require_resolved_commit(head_info["commit"], "HEAD"), 1388 user=self._config.user, 1389 message=message, 1390 db=self._db, 1391 missing_commits=head.get_shallow_commits(), 1392 ) 1393 head.update_local_ref(head_info["branch"], new_commit, kind="branch") 1394 return self.status() 1395 1396 def checkout( 1397 self, revision: Annotated[Ref | str, "Revision to check out."], *, remote: bool = False 1398 ) -> StatusPayload: 1399 """Check out a different revision. 1400 1401 If the revision resolves to a local branch, HEAD stays attached to that branch. 1402 Other revisions detach HEAD at the resolved commit. 1403 """ 1404 head = _head_ops(self) 1405 with head.lock(): 1406 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote) 1407 if ( 1408 not remote 1409 and isinstance(revision, str) 1410 and not revision.startswith(("@", "HEAD")) 1411 and not re.match(r"^(?:commit:)?[0-9a-f]{64}$", revision) 1412 ): 1413 head.write_attached_head(revision) 1414 else: 1415 head.write_detached_head(_require_resolved_commit(commit_ref, revision)) 1416 return self.status() 1417 1418 def merge( 1419 self, 1420 revision: Annotated[Ref | str, "Revision to merge into the current branch."], 1421 ff_only: Annotated[bool, "Whether to only allow fast-forward merges."] = True, 1422 *, 1423 remote: bool = False, 1424 ) -> StatusPayload: 1425 """Merge a revision into the current HEAD.""" 1426 head = _head_ops(self) 1427 with head.lock(): 1428 head_info = head.get_head() 1429 if head_info["branch"] is None: 1430 raise DmlRepoError("Cannot merge when HEAD is detached") 1431 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote) 1432 commit_ref = _require_resolved_commit(commit_ref, revision) 1433 new_commit = CommitOps().merge( 1434 head_info["commit"], 1435 commit_ref, 1436 user=self._config.user, 1437 ff_only=ff_only, 1438 db=self._db, 1439 missing_commits=head.get_shallow_commits(), 1440 ) 1441 head.update_local_ref(head_info["branch"], new_commit) 1442 return self.status() 1443 1444 def rebase( 1445 self, revision: Annotated[Ref | str, "Revision to rebase the current branch onto."], *, remote: bool = False 1446 ) -> StatusPayload: 1447 """Rebase the current HEAD onto a different revision.""" 1448 head = _head_ops(self) 1449 with head.lock(): 1450 head_info = head.get_head() 1451 if head_info["branch"] is None: 1452 raise DmlRepoError("Cannot rebase when HEAD is detached") 1453 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote) 1454 new_commit = CommitOps().rebase( 1455 _require_resolved_commit(head_info["commit"], "HEAD"), 1456 _require_resolved_commit(commit_ref, revision), 1457 user=self._config.user, 1458 db=self._db, 1459 missing_commits=head.get_shallow_commits(), 1460 ) 1461 head.update_local_ref(head_info["branch"], new_commit) 1462 return self.status() 1463 1464 def fetch( 1465 self, 1466 revision: Annotated[str | None, "Branch or @tag to fetch."] = None, 1467 /, 1468 *, 1469 dep: Annotated[str | None, "Named dependency endpoint to fetch from."] = None, 1470 depth: Annotated[int | None, "Positive number of commit-history generations to fetch."] = None, 1471 unshallow: Annotated[bool, "Fetch all history through existing shallow boundaries."] = False, 1472 ) -> None: 1473 """Fetch one branch or tag from remote.root or a named dependency.""" 1474 _validate_history_options(depth, unshallow) 1475 selector = revision or self._config.default.branch_name 1476 kind = "tag" if selector.startswith("@") else "branch" 1477 name = selector[1:] if kind == "tag" else selector 1478 head = _head_ops(self) 1479 if dep is None: 1480 remote_ops = _remote_ops(self) 1481 else: 1482 config = head.get_dependency_config(dep) 1483 remote_ops = Remote( 1484 config["root"], 1485 n_workers=self._config.remote.fetch_workers, 1486 client=_require_s3_client(self), 1487 prune_age_seconds=self._config.remote.prune_age_seconds, 1488 ) 1489 materialized = remote_ops.get_project_commit_ref( 1490 kind, 1491 name, 1492 db=self._db, 1493 depth=depth, 1494 unshallow=unshallow, 1495 ) 1496 if materialized is None: 1497 raise DmlRepoError(f"Remote {kind} ref not found: {selector}") 1498 commit, available, omitted = materialized 1499 with head.lock(): 1500 _publish_shallow_state(head, available, omitted) 1501 if dep is None: 1502 head.update_remote_tracking_ref(name, commit, kind=kind) 1503 else: 1504 head.update_dependency_ref(dep, name, commit, kind=kind) 1505 1506 def pull( 1507 self, 1508 ff_only: Annotated[bool, "Whether to only allow fast-forward merges."] = True, 1509 *, 1510 depth: Annotated[int | None, "Positive number of commit-history generations to fetch."] = None, 1511 ) -> StatusPayload: 1512 """Fetch and merge the current branch's configured upstream.""" 1513 _validate_history_options(depth) 1514 head = _head_ops(self) 1515 head_info = head.get_head() 1516 if head_info["branch"] is None: 1517 raise DmlRepoError("Cannot pull when HEAD is detached") 1518 upstream = head.get_upstream(head_info["branch"]) 1519 if upstream is None: 1520 raise DmlRepoError(f"Cannot pull untracked branch: {head_info['branch']}") 1521 self.fetch(upstream["branch"], depth=depth) 1522 self.merge(upstream["branch"], ff_only=ff_only, remote=True) 1523 return self.status() 1524 1525 def push( 1526 self, 1527 *, 1528 revision: Annotated[Ref | str, "Local revision to publish; defaults to HEAD."] = "HEAD", 1529 force: Annotated[bool, "Overwrite a remote branch or tag without publication checks."] = False, 1530 ) -> None: 1531 """Publish a local revision as a tag or branch on the configured remote.""" 1532 head = _head_ops(self) 1533 commit_ref = _require_resolved_commit(resolve_rev(head, revision, db=self._db), revision) 1534 if isinstance(revision, str) and revision.startswith("@"): 1535 _remote_ops(self).put_ref( 1536 commit_ref, 1537 kind="tag", 1538 name=revision[1:], 1539 db=self._db, 1540 force=force, 1541 missing_commits=head.get_shallow_commits(), 1542 ) 1543 return 1544 named_branch = ( 1545 isinstance(revision, str) 1546 and not revision.startswith("HEAD") 1547 and re.fullmatch(r"(?:commit:)?[0-9a-f]{64}", revision) is None 1548 ) 1549 branch = revision if named_branch and isinstance(revision, str) else head.get_head()["branch"] 1550 if branch is None: 1551 raise DmlRepoError("Cannot push an unnamed revision when HEAD is detached") 1552 upstream = head.get_upstream(branch) 1553 if upstream is None: 1554 upstream = {"branch": branch} 1555 set_upstream = True 1556 else: 1557 set_upstream = False 1558 remote = _remote_ops(self) 1559 remote.put_ref( 1560 commit_ref, 1561 kind="branch", 1562 name=upstream["branch"], 1563 db=self._db, 1564 force=force, 1565 missing_commits=head.get_shallow_commits(), 1566 ) 1567 if set_upstream: 1568 with head.lock(): 1569 head.set_upstream(branch, upstream["branch"]) 1570 1571 def gc( 1572 self, 1573 *, 1574 remote: Annotated[bool, "Garbage-collect configured remote state instead of local objects."] = False, 1575 ) -> LocalGCSummary | RemoteGCSummary: 1576 """Garbage-collect unreachable local or configured remote state.""" 1577 if not remote: 1578 return _local_gc(self) 1579 return cast(RemoteGCSummary, _remote_ops(self).gc()) 1580 1581 @property 1582 def branch(self) -> Annotated[_BranchNamespace, "Branch inspection and lifecycle commands."]: 1583 """Expose branch inspection and lifecycle commands.""" 1584 return _BranchNamespace(self) 1585 1586 @property 1587 def dep(self) -> Annotated[_DependencyNamespace, "Import-only dependency lifecycle commands."]: 1588 """Expose import-only dependency lifecycle commands.""" 1589 return _DependencyNamespace(self) 1590 1591 @property 1592 def cache(self) -> Annotated[_CacheNamespace, "Remote execution cache inspection and control commands."]: 1593 """Expose remote execution cache inspection and control commands.""" 1594 return _CacheNamespace(self) 1595 1596 @property 1597 def tag(self) -> Annotated[_TagNamespace, "Tag inspection and lifecycle commands."]: 1598 """Expose tag inspection and lifecycle commands.""" 1599 return _TagNamespace(self) 1600 1601 @property 1602 def config(self) -> Annotated[_ConfigNamespace, "Configuration commands."]: 1603 """Expose configuration commands.""" 1604 return _ConfigNamespace(self) 1605 1606 @property 1607 def runtime(self) -> Annotated[_RuntimeNamespace, "Runtime mutation and execution-state inspection commands."]: 1608 """Expose runtime mutation and execution-state inspection commands.""" 1609 return _RuntimeNamespace(self) 1610 1611 @property 1612 def dag(self) -> Annotated[_DagNamespace, "Committed DAG inspection commands."]: 1613 """Expose committed DAG inspection commands.""" 1614 return _DagNamespace(self) 1615 1616 @property 1617 def skills(self) -> Annotated[_SkillsNamespace, "Bundled agent-guidance exports."]: 1618 """Expose bundled agent-guidance exports.""" 1619 return _SkillsNamespace(self)
1082 def __init__( 1083 self, 1084 project_home: Annotated[str | None, "Project root containing the .dml repository."] = None, 1085 *, 1086 db_path: Annotated[str | None, "Override path to the LMDB database."] = None, 1087 db_map_size_headroom: Annotated[int | None, "Extra LMDB map size headroom in bytes."] = None, 1088 db_map_size_max: Annotated[int | None, "Maximum LMDB map size in bytes."] = None, 1089 default_branch_name: Annotated[str | None, "Default branch name for attached HEAD operations."] = None, 1090 remote_root: Annotated[str | None, "Remote storage root URI."] = None, 1091 remote_prune_age_seconds: Annotated[int | None, "Remote GC prune age in seconds."] = None, 1092 remote_fetch_workers: Annotated[int | None, "Number of concurrent remote fetch workers."] = None, 1093 user: Annotated[str | None, "User name recorded in commits and runtime actions."] = None, 1094 config_home: Annotated[str | None, "Override config directory path."] = None, 1095 ): 1096 """Create a DaggerML session bound to one repository and config context.""" 1097 self._init_from_config_vars( 1098 _python_config_vars_to_canonical( 1099 project_home=project_home, 1100 db_path=db_path, 1101 db_map_size_headroom=db_map_size_headroom, 1102 db_map_size_max=db_map_size_max, 1103 default_branch_name=default_branch_name, 1104 remote_root=remote_root, 1105 remote_prune_age_seconds=remote_prune_age_seconds, 1106 remote_fetch_workers=remote_fetch_workers, 1107 user=user, 1108 config_home=config_home, 1109 ) 1110 )
Create a DaggerML session bound to one repository and config context.
1112 @classmethod 1113 def from_config_vars( 1114 cls, 1115 config_vars: Annotated[dict[str, object], "Flattened canonical config-var mapping."] | None = None, 1116 ) -> "Dml": 1117 """Create a DaggerML session from flattened canonical config vars.""" 1118 dml = cls.__new__(cls) 1119 dml._init_from_config_vars(config_vars or {}) 1120 return dml
Create a DaggerML session from flattened canonical config vars.
1122 @classmethod 1123 def init( 1124 cls, 1125 project_home: Annotated[str, "Directory where the repository should be initialized."] = ".", 1126 *, 1127 db_path: Annotated[str | None, "Override path to the LMDB database."] = None, 1128 db_map_size_headroom: Annotated[int | None, "Extra LMDB map size headroom in bytes."] = None, 1129 db_map_size_max: Annotated[int | None, "Maximum LMDB map size in bytes."] = None, 1130 default_branch_name: Annotated[str | None, "Default branch name for attached HEAD operations."] = None, 1131 remote_root: Annotated[str | None, "Remote storage root URI."] = None, 1132 remote_prune_age_seconds: Annotated[int | None, "Remote GC prune age in seconds."] = None, 1133 remote_fetch_workers: Annotated[int | None, "Number of concurrent remote fetch workers."] = None, 1134 user: Annotated[str | None, "User name recorded in commits and runtime actions."] = None, 1135 config_home: Annotated[str | None, "Override config directory path."] = None, 1136 branch: Annotated[str | None, "Initial branch name."] = None, 1137 ) -> "Dml": 1138 """Initialize a repository with an unborn attached HEAD.""" 1139 config = Config.init(project_home, remote_root=remote_root) 1140 dml = cls.from_config_vars( 1141 _python_config_vars_to_canonical( 1142 project_home=config.project_home, 1143 db_path=db_path, 1144 db_map_size_headroom=db_map_size_headroom, 1145 db_map_size_max=db_map_size_max, 1146 default_branch_name=default_branch_name, 1147 remote_root=remote_root, 1148 remote_prune_age_seconds=remote_prune_age_seconds, 1149 remote_fetch_workers=remote_fetch_workers, 1150 user=user, 1151 config_home=config_home, 1152 ) 1153 ) 1154 head = Head(config.project_home) 1155 branch = branch or dml._config.default.branch_name 1156 with head.lock(): 1157 try: 1158 head.get_head() 1159 except (DmlRepoError, FileNotFoundError): 1160 dml._db.init() 1161 head.init(None, branch) 1162 return dml
Initialize a repository with an unborn attached HEAD.
1164 @classmethod 1165 def clone( 1166 cls, 1167 revision: Annotated[Ref | str | None, "Optional branch, tag, or commit revision."] = None, 1168 /, 1169 *, 1170 project_home: Annotated[str, "Directory where the repository should be cloned."] = ".", 1171 db_path: Annotated[str | None, "Override path to the LMDB database."] = None, 1172 db_map_size_headroom: Annotated[int | None, "Extra LMDB map size headroom in bytes."] = None, 1173 db_map_size_max: Annotated[int | None, "Maximum LMDB map size in bytes."] = None, 1174 default_branch_name: Annotated[str | None, "Default branch name for attached HEAD operations."] = None, 1175 remote_root: Annotated[str | None, "Remote storage root URI."] = None, 1176 remote_prune_age_seconds: Annotated[int | None, "Remote GC prune age in seconds."] = None, 1177 remote_fetch_workers: Annotated[int | None, "Number of concurrent remote fetch workers."] = None, 1178 user: Annotated[str | None, "User name recorded in commits and runtime actions."] = None, 1179 config_home: Annotated[str | None, "Override config directory path."] = None, 1180 depth: Annotated[int | None, "Positive number of commit-history generations to fetch."] = None, 1181 ) -> "Dml": 1182 """Clone one revision from the configured remote root.""" 1183 _validate_history_options(depth) 1184 Path(project_home).mkdir(parents=True, exist_ok=True) 1185 config = Config.init(project_home, remote_root=remote_root) 1186 dml = cls.from_config_vars( 1187 _python_config_vars_to_canonical( 1188 project_home=config.project_home, 1189 db_path=db_path, 1190 db_map_size_headroom=db_map_size_headroom, 1191 db_map_size_max=db_map_size_max, 1192 default_branch_name=default_branch_name, 1193 remote_root=remote_root, 1194 remote_prune_age_seconds=remote_prune_age_seconds, 1195 remote_fetch_workers=remote_fetch_workers, 1196 user=user, 1197 config_home=config_home, 1198 ) 1199 ) 1200 _require_remote_root(dml) 1201 selected = revision or dml._config.default.branch_name 1202 exact = selected if isinstance(selected, Ref) else None 1203 if isinstance(selected, str) and re.match(r"^(?:commit:)?[0-9a-f]{64}$", selected): 1204 exact = Ref(selected if selected.startswith("commit:") else f"commit:{selected}") 1205 branch = ( 1206 selected 1207 if exact is None and isinstance(selected, str) and not selected.startswith(("@", "HEAD")) 1208 else None 1209 ) 1210 initial_branch = branch or dml._config.default.branch_name 1211 head = Head(config.project_home) 1212 with head.lock(): 1213 try: 1214 head.get_head() 1215 except (DmlRepoError, FileNotFoundError): 1216 dml._db.init() 1217 head.init(None, initial_branch) 1218 else: 1219 raise DmlRepoError(f"Cannot clone into an initialized repository: {dml._config.project_home}") 1220 if exact is not None: 1221 commit, available, omitted = _remote_ops(dml).materialize_project_commit_ref(exact, dml._db, depth=depth) 1222 with head.lock(): 1223 _publish_shallow_state(head, available, omitted) 1224 head.write_detached_head(commit) 1225 return dml 1226 dml.fetch(selected if isinstance(selected, str) else None, depth=depth) 1227 kind = "tag" if isinstance(selected, str) and selected.startswith("@") else "branch" 1228 name = selected[1:] if kind == "tag" and isinstance(selected, str) else selected 1229 assert isinstance(name, str) 1230 with head.lock(): 1231 commit = head.get_remote_tracking_ref(name, kind=kind) 1232 if kind == "tag": 1233 head.write_detached_head(commit) 1234 else: 1235 head.update_local_ref(name, commit, kind="branch") 1236 head.write_attached_head(name) 1237 head.set_upstream(name, name) 1238 return dml
Clone one revision from the configured remote root.
1240 def status(self) -> StatusPayload: 1241 """Return branch, commit, and open-runtime status for this repository.""" 1242 head = _head_ops(self) 1243 head_info = head.get_head() 1244 ahead = behind = None 1245 upstream = head.get_upstream(head_info["branch"]) if head_info["branch"] is not None else None 1246 if upstream is not None: 1247 try: 1248 upstream_ref = head.get_remote_tracking_ref(upstream["branch"]) 1249 except DmlRepoError: 1250 pass 1251 else: 1252 if head_info["commit"] is not None: 1253 try: 1254 ahead, behind = CommitOps().ahead_behind( 1255 head_info["commit"], 1256 upstream_ref, 1257 db=self._db, 1258 missing_commits=head.get_shallow_commits(), 1259 ) 1260 except ShallowHistoryError: 1261 pass 1262 with self._db.tx(readonly=True) as txn: 1263 num_indexes = 0 1264 for namespace in ("index", "frozenindex"): 1265 try: 1266 num_indexes += sum(1 for _ in txn.iter(namespace)) 1267 except DmlDbKeyNotFoundError: 1268 pass 1269 return { 1270 "mode": head_info["mode"], 1271 "branch": head_info["branch"], 1272 "commit": head_info["commit"], 1273 "branches": head.list_local_refs(kind="branch"), 1274 "upstream": upstream["branch"] if upstream is not None else None, 1275 "num_indexes": num_indexes, 1276 "ahead": ahead, 1277 "behind": behind, 1278 }
Return branch, commit, and open-runtime status for this repository.
1280 def log( 1281 self, 1282 revision: Annotated[Ref | str, "Revision to start the log from."] = "HEAD", 1283 limit: Annotated[int, "Maximum number of commits to return."] = 10, 1284 *, 1285 remote: Annotated[bool, "Resolve from fetched remote tracking refs."] = False, 1286 dep: Annotated[str | None, "Resolve from a fetched dependency."] = None, 1287 ) -> LogPayload: 1288 """Return commit history starting from one revision.""" 1289 commit_ref = resolve_rev(_head_ops(self), revision, db=self._db, remote=remote, dep=dep) 1290 commit_ref = _require_resolved_commit(commit_ref, revision) 1291 commits, truncated = CommitOps().log_with_truncation( 1292 commit_ref, 1293 limit=limit, 1294 db=self._db, 1295 missing_commits=_head_ops(self).get_shallow_commits(), 1296 ) 1297 return {"commits": commits, "truncated": truncated}
Return commit history starting from one revision.
1299 def show( 1300 self, 1301 revision: Annotated[Ref | str, "Revision to describe."] = "HEAD", 1302 *, 1303 remote: bool = False, 1304 dep: str | None = None, 1305 ) -> CommitFullDescription: 1306 """Return a full commit description for one revision.""" 1307 commit_ref = resolve_rev(_head_ops(self), revision, db=self._db, remote=remote, dep=dep) 1308 commit_ref = _require_resolved_commit(commit_ref, revision) 1309 return CommitOps().show( 1310 commit_ref, 1311 db=self._db, 1312 missing_commits=_head_ops(self).get_shallow_commits(), 1313 )
Return a full commit description for one revision.
1315 def diff( 1316 self, 1317 revision: Annotated[Ref | str, "Revision to diff."] = "HEAD", 1318 relative_to: Annotated[Ref | str | None, "Optional base revision. Defaults to the commit parent."] = None, 1319 *, 1320 remote: Annotated[bool, "Resolve the primary revision from remote tracking."] = False, 1321 dep: Annotated[str | None, "Resolve the primary revision from a dependency."] = None, 1322 ) -> CommitDiffPayload: 1323 """Return DAG-level changes for one revision.""" 1324 head = _head_ops(self) 1325 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote, dep=dep) 1326 commit_ref = _require_resolved_commit(commit_ref, revision) 1327 if relative_to is None: 1328 return CommitOps().diff( 1329 commit_ref, 1330 db=self._db, 1331 missing_commits=head.get_shallow_commits(), 1332 ) 1333 rel_to_commit = resolve_rev(head, relative_to, db=self._db) 1334 rel_to_commit = _require_resolved_commit(rel_to_commit, relative_to) 1335 return CommitOps().diff( 1336 commit_ref, 1337 rel_to_commit, 1338 db=self._db, 1339 missing_commits=head.get_shallow_commits(), 1340 )
Return DAG-level changes for one revision.
1342 def rev_parse( 1343 self, 1344 revision: Annotated[str, "Revision expression to resolve."], 1345 *, 1346 remote: bool = False, 1347 dep: str | None = None, 1348 ) -> RevisionPayload: 1349 """Resolve a revision expression into a commit and ref metadata.""" 1350 head = _head_ops(self) 1351 branch = tag = None 1352 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote, dep=dep) 1353 if revision.startswith("HEAD"): 1354 kind = "head" 1355 elif re.match(r"^(?:commit:)?[0-9a-f]{64}$", revision): 1356 kind = "commit" 1357 else: 1358 kind = "ref" 1359 if kind == "ref": 1360 branch, tag = (None, revision[1:]) if revision.startswith("@") else (revision, None) 1361 return { 1362 "input": revision, 1363 "uri": None, 1364 "kind": kind, 1365 "commit": commit_ref, 1366 "branch": branch, 1367 "tag": tag, 1368 }
Resolve a revision expression into a commit and ref metadata.
1370 def revert( 1371 self, 1372 revision: Annotated[Ref | str, "Revision whose changes should be reverted."], 1373 message: Annotated[str | None, "Optional commit message for the revert commit."] = None, 1374 *, 1375 remote: bool = False, 1376 ) -> StatusPayload: 1377 """Revert the changes introduced by one revision.""" 1378 head = _head_ops(self) 1379 with head.lock(): 1380 head_info = head.get_head() 1381 if head_info["branch"] is None: 1382 raise DmlRepoError("Cannot revert when HEAD is detached") 1383 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote) 1384 commit_ref = _require_resolved_commit(commit_ref, revision) 1385 new_commit = CommitOps().revert( 1386 commit_ref, 1387 _require_resolved_commit(head_info["commit"], "HEAD"), 1388 user=self._config.user, 1389 message=message, 1390 db=self._db, 1391 missing_commits=head.get_shallow_commits(), 1392 ) 1393 head.update_local_ref(head_info["branch"], new_commit, kind="branch") 1394 return self.status()
Revert the changes introduced by one revision.
1396 def checkout( 1397 self, revision: Annotated[Ref | str, "Revision to check out."], *, remote: bool = False 1398 ) -> StatusPayload: 1399 """Check out a different revision. 1400 1401 If the revision resolves to a local branch, HEAD stays attached to that branch. 1402 Other revisions detach HEAD at the resolved commit. 1403 """ 1404 head = _head_ops(self) 1405 with head.lock(): 1406 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote) 1407 if ( 1408 not remote 1409 and isinstance(revision, str) 1410 and not revision.startswith(("@", "HEAD")) 1411 and not re.match(r"^(?:commit:)?[0-9a-f]{64}$", revision) 1412 ): 1413 head.write_attached_head(revision) 1414 else: 1415 head.write_detached_head(_require_resolved_commit(commit_ref, revision)) 1416 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.
1418 def merge( 1419 self, 1420 revision: Annotated[Ref | str, "Revision to merge into the current branch."], 1421 ff_only: Annotated[bool, "Whether to only allow fast-forward merges."] = True, 1422 *, 1423 remote: bool = False, 1424 ) -> StatusPayload: 1425 """Merge a revision into the current HEAD.""" 1426 head = _head_ops(self) 1427 with head.lock(): 1428 head_info = head.get_head() 1429 if head_info["branch"] is None: 1430 raise DmlRepoError("Cannot merge when HEAD is detached") 1431 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote) 1432 commit_ref = _require_resolved_commit(commit_ref, revision) 1433 new_commit = CommitOps().merge( 1434 head_info["commit"], 1435 commit_ref, 1436 user=self._config.user, 1437 ff_only=ff_only, 1438 db=self._db, 1439 missing_commits=head.get_shallow_commits(), 1440 ) 1441 head.update_local_ref(head_info["branch"], new_commit) 1442 return self.status()
Merge a revision into the current HEAD.
1444 def rebase( 1445 self, revision: Annotated[Ref | str, "Revision to rebase the current branch onto."], *, remote: bool = False 1446 ) -> StatusPayload: 1447 """Rebase the current HEAD onto a different revision.""" 1448 head = _head_ops(self) 1449 with head.lock(): 1450 head_info = head.get_head() 1451 if head_info["branch"] is None: 1452 raise DmlRepoError("Cannot rebase when HEAD is detached") 1453 commit_ref = resolve_rev(head, revision, db=self._db, remote=remote) 1454 new_commit = CommitOps().rebase( 1455 _require_resolved_commit(head_info["commit"], "HEAD"), 1456 _require_resolved_commit(commit_ref, revision), 1457 user=self._config.user, 1458 db=self._db, 1459 missing_commits=head.get_shallow_commits(), 1460 ) 1461 head.update_local_ref(head_info["branch"], new_commit) 1462 return self.status()
Rebase the current HEAD onto a different revision.
1464 def fetch( 1465 self, 1466 revision: Annotated[str | None, "Branch or @tag to fetch."] = None, 1467 /, 1468 *, 1469 dep: Annotated[str | None, "Named dependency endpoint to fetch from."] = None, 1470 depth: Annotated[int | None, "Positive number of commit-history generations to fetch."] = None, 1471 unshallow: Annotated[bool, "Fetch all history through existing shallow boundaries."] = False, 1472 ) -> None: 1473 """Fetch one branch or tag from remote.root or a named dependency.""" 1474 _validate_history_options(depth, unshallow) 1475 selector = revision or self._config.default.branch_name 1476 kind = "tag" if selector.startswith("@") else "branch" 1477 name = selector[1:] if kind == "tag" else selector 1478 head = _head_ops(self) 1479 if dep is None: 1480 remote_ops = _remote_ops(self) 1481 else: 1482 config = head.get_dependency_config(dep) 1483 remote_ops = Remote( 1484 config["root"], 1485 n_workers=self._config.remote.fetch_workers, 1486 client=_require_s3_client(self), 1487 prune_age_seconds=self._config.remote.prune_age_seconds, 1488 ) 1489 materialized = remote_ops.get_project_commit_ref( 1490 kind, 1491 name, 1492 db=self._db, 1493 depth=depth, 1494 unshallow=unshallow, 1495 ) 1496 if materialized is None: 1497 raise DmlRepoError(f"Remote {kind} ref not found: {selector}") 1498 commit, available, omitted = materialized 1499 with head.lock(): 1500 _publish_shallow_state(head, available, omitted) 1501 if dep is None: 1502 head.update_remote_tracking_ref(name, commit, kind=kind) 1503 else: 1504 head.update_dependency_ref(dep, name, commit, kind=kind)
Fetch one branch or tag from remote.root or a named dependency.
1506 def pull( 1507 self, 1508 ff_only: Annotated[bool, "Whether to only allow fast-forward merges."] = True, 1509 *, 1510 depth: Annotated[int | None, "Positive number of commit-history generations to fetch."] = None, 1511 ) -> StatusPayload: 1512 """Fetch and merge the current branch's configured upstream.""" 1513 _validate_history_options(depth) 1514 head = _head_ops(self) 1515 head_info = head.get_head() 1516 if head_info["branch"] is None: 1517 raise DmlRepoError("Cannot pull when HEAD is detached") 1518 upstream = head.get_upstream(head_info["branch"]) 1519 if upstream is None: 1520 raise DmlRepoError(f"Cannot pull untracked branch: {head_info['branch']}") 1521 self.fetch(upstream["branch"], depth=depth) 1522 self.merge(upstream["branch"], ff_only=ff_only, remote=True) 1523 return self.status()
Fetch and merge the current branch's configured upstream.
1525 def push( 1526 self, 1527 *, 1528 revision: Annotated[Ref | str, "Local revision to publish; defaults to HEAD."] = "HEAD", 1529 force: Annotated[bool, "Overwrite a remote branch or tag without publication checks."] = False, 1530 ) -> None: 1531 """Publish a local revision as a tag or branch on the configured remote.""" 1532 head = _head_ops(self) 1533 commit_ref = _require_resolved_commit(resolve_rev(head, revision, db=self._db), revision) 1534 if isinstance(revision, str) and revision.startswith("@"): 1535 _remote_ops(self).put_ref( 1536 commit_ref, 1537 kind="tag", 1538 name=revision[1:], 1539 db=self._db, 1540 force=force, 1541 missing_commits=head.get_shallow_commits(), 1542 ) 1543 return 1544 named_branch = ( 1545 isinstance(revision, str) 1546 and not revision.startswith("HEAD") 1547 and re.fullmatch(r"(?:commit:)?[0-9a-f]{64}", revision) is None 1548 ) 1549 branch = revision if named_branch and isinstance(revision, str) else head.get_head()["branch"] 1550 if branch is None: 1551 raise DmlRepoError("Cannot push an unnamed revision when HEAD is detached") 1552 upstream = head.get_upstream(branch) 1553 if upstream is None: 1554 upstream = {"branch": branch} 1555 set_upstream = True 1556 else: 1557 set_upstream = False 1558 remote = _remote_ops(self) 1559 remote.put_ref( 1560 commit_ref, 1561 kind="branch", 1562 name=upstream["branch"], 1563 db=self._db, 1564 force=force, 1565 missing_commits=head.get_shallow_commits(), 1566 ) 1567 if set_upstream: 1568 with head.lock(): 1569 head.set_upstream(branch, upstream["branch"])
Publish a local revision as a tag or branch on the configured remote.
1571 def gc( 1572 self, 1573 *, 1574 remote: Annotated[bool, "Garbage-collect configured remote state instead of local objects."] = False, 1575 ) -> LocalGCSummary | RemoteGCSummary: 1576 """Garbage-collect unreachable local or configured remote state.""" 1577 if not remote: 1578 return _local_gc(self) 1579 return cast(RemoteGCSummary, _remote_ops(self).gc())
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.