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