| 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 | k: v for k, v in subs.items() if k in self.dependencies and k != v |
| 796 | } |
| 797 | extras = _extra_args(type(self)) # type: ignore[arg-type] |
| 798 | extra_kwargs = { |
| 799 | name: getattr(self, name) for name in extras if name not in {"key", "func"} |
| 800 | } |
| 801 | if subs_filtered: |
| 802 | new_args = tuple( |
| 803 | ( |
| 804 | a.substitute(subs_filtered) |
| 805 | if isinstance(a, (GraphNode, TaskRef)) |
| 806 | else a |
| 807 | ) |
| 808 | for a in self.args |
| 809 | ) |
| 810 | new_kwargs = { |
| 811 | k: ( |
| 812 | v.substitute(subs_filtered) |
| 813 | if isinstance(v, (GraphNode, TaskRef)) |
| 814 | else v |
| 815 | ) |
| 816 | for k, v in self.kwargs.items() |
| 817 | } |
| 818 | return type(self)( |
| 819 | key or self.key, |
| 820 | self.func, |
| 821 | *new_args, |
| 822 | **new_kwargs, # type: ignore[arg-type] |
| 823 | **extra_kwargs, |
| 824 | ) |
| 825 | elif key is None or key == self.key: |
| 826 | return self |
| 827 | else: |
| 828 | # Rename |
| 829 | return type(self)( |
| 830 | key, |
| 831 | self.func, |
| 832 | *self.args, |
| 833 | **self.kwargs, |
| 834 | **extra_kwargs, |
| 835 | ) |
| 836 | |
| 837 | |
| 838 | class NestedContainer(Task, Iterable): |