(
self, subs: dict[KeyType, KeyType | GraphNode], key: KeyType | None = None
)
| 535 | return Alias(self.key, self.target) |
| 536 | |
| 537 | def substitute( |
| 538 | self, subs: dict[KeyType, KeyType | GraphNode], key: KeyType | None = None |
| 539 | ) -> GraphNode: |
| 540 | if self.key in subs or self.target in subs: |
| 541 | sub_key = subs.get(self.key, self.key) |
| 542 | val = subs.get(self.target, self.target) |
| 543 | if sub_key == self.key and val == self.target: |
| 544 | return self |
| 545 | if isinstance(val, (GraphNode, TaskRef)): |
| 546 | return val.substitute({}, key=key) |
| 547 | if key is None and isinstance(sub_key, GraphNode): |
| 548 | raise RuntimeError( |
| 549 | f"Invalid substitution encountered {self.key!r} -> {sub_key}" |
| 550 | ) |
| 551 | return Alias(key or sub_key, val) # type: ignore [arg-type] |
| 552 | return self |
| 553 | |
| 554 | def __dask_tokenize__(self): |
| 555 | return (type(self).__name__, self.key, self.target) |
nothing calls this directly
no test coverage detected