(
self, subs: dict[KeyType, KeyType | GraphNode], key: KeyType | None = None
)
| 526 | return Alias(self.key, self.target) |
| 527 | |
| 528 | def substitute( |
| 529 | self, subs: dict[KeyType, KeyType | GraphNode], key: KeyType | None = None |
| 530 | ) -> GraphNode: |
| 531 | if self.key in subs or self.target in subs: |
| 532 | sub_key = subs.get(self.key, self.key) |
| 533 | val = subs.get(self.target, self.target) |
| 534 | if sub_key == self.key and val == self.target: |
| 535 | return self |
| 536 | if isinstance(val, (GraphNode, TaskRef)): |
| 537 | return val.substitute({}, key=key) |
| 538 | if key is None and isinstance(sub_key, GraphNode): |
| 539 | raise RuntimeError( |
| 540 | f"Invalid substitution encountered {self.key!r} -> {sub_key}" |
| 541 | ) |
| 542 | return Alias(key or sub_key, val) # type: ignore |
| 543 | return self |
| 544 | |
| 545 | def __dask_tokenize__(self): |
| 546 | return (type(self).__name__, self.key, self.target) |
nothing calls this directly
no test coverage detected