| 789 | return self._is_coro |
| 790 | |
| 791 | def substitute( |
| 792 | self, subs: dict[KeyType, KeyType | GraphNode], key: KeyType | None = None |
| 793 | ) -> Task: |
| 794 | subs_filtered = { |
| 795 | # subs can be as large as the whole graph; do not iterate over it! |
| 796 | k: v |
| 797 | for k in (self.dependencies & subs.keys()) |
| 798 | if (v := subs[k]) != k |
| 799 | } |
| 800 | extras = _extra_args(type(self)) # type: ignore[arg-type] |
| 801 | extra_kwargs = { |
| 802 | name: getattr(self, name) for name in extras if name not in {"key", "func"} |
| 803 | } |
| 804 | if subs_filtered: |
| 805 | new_args = tuple( |
| 806 | ( |
| 807 | a.substitute(subs_filtered) |
| 808 | if isinstance(a, (GraphNode, TaskRef)) |
| 809 | else a |
| 810 | ) |
| 811 | for a in self.args |
| 812 | ) |
| 813 | new_kwargs = { |
| 814 | k: ( |
| 815 | v.substitute(subs_filtered) |
| 816 | if isinstance(v, (GraphNode, TaskRef)) |
| 817 | else v |
| 818 | ) |
| 819 | for k, v in self.kwargs.items() |
| 820 | } |
| 821 | return type(self)( |
| 822 | key or self.key, |
| 823 | self.func, |
| 824 | *new_args, |
| 825 | **new_kwargs, # type: ignore[arg-type] |
| 826 | **extra_kwargs, |
| 827 | ) |
| 828 | elif key is None or key == self.key: |
| 829 | return self |
| 830 | else: |
| 831 | # Rename |
| 832 | return type(self)( |
| 833 | key, |
| 834 | self.func, |
| 835 | *self.args, |
| 836 | **self.kwargs, |
| 837 | **extra_kwargs, |
| 838 | ) |
| 839 | |
| 840 | |
| 841 | class NestedContainer(Task, Iterable): |