(
self, subs: dict[KeyType, KeyType | GraphNode], key: KeyType | None = None
)
| 965 | return f"Dict({values})" |
| 966 | |
| 967 | def substitute( |
| 968 | self, subs: dict[KeyType, KeyType | GraphNode], key: KeyType | None = None |
| 969 | ) -> Dict: |
| 970 | subs_filtered = { |
| 971 | # subs can be as large as the whole graph; do not iterate over it! |
| 972 | k: v |
| 973 | for k in (self.dependencies & subs.keys()) |
| 974 | if (v := subs[k]) != k |
| 975 | } |
| 976 | if not subs_filtered: |
| 977 | return self |
| 978 | |
| 979 | new_args = [] |
| 980 | for arg in self.args: |
| 981 | new_arg = ( |
| 982 | arg.substitute(subs_filtered) |
| 983 | if isinstance(arg, (GraphNode, TaskRef)) |
| 984 | else arg |
| 985 | ) |
| 986 | new_args.append(new_arg) |
| 987 | return type(self)(new_args) |
| 988 | |
| 989 | def __iter__(self): |
| 990 | yield from self.args[::2] |
nothing calls this directly
no test coverage detected