| 780 | return self._is_coro |
| 781 | |
| 782 | def substitute( |
| 783 | self, subs: dict[KeyType, KeyType | GraphNode], key: KeyType | None = None |
| 784 | ) -> Task: |
| 785 | subs_filtered = { |
| 786 | k: v for k, v in subs.items() if k in self.dependencies and k != v |
| 787 | } |
| 788 | extras = _extra_args(type(self)) # type: ignore |
| 789 | extra_kwargs = { |
| 790 | name: getattr(self, name) for name in extras if name not in {"key", "func"} |
| 791 | } |
| 792 | if subs_filtered: |
| 793 | new_args = tuple( |
| 794 | ( |
| 795 | a.substitute(subs_filtered) |
| 796 | if isinstance(a, (GraphNode, TaskRef)) |
| 797 | else a |
| 798 | ) |
| 799 | for a in self.args |
| 800 | ) |
| 801 | new_kwargs = { |
| 802 | k: ( |
| 803 | v.substitute(subs_filtered) |
| 804 | if isinstance(v, (GraphNode, TaskRef)) |
| 805 | else v |
| 806 | ) |
| 807 | for k, v in self.kwargs.items() |
| 808 | } |
| 809 | return type(self)( |
| 810 | key or self.key, |
| 811 | self.func, |
| 812 | *new_args, |
| 813 | **new_kwargs, # type: ignore[arg-type] |
| 814 | **extra_kwargs, |
| 815 | ) |
| 816 | elif key is None or key == self.key: |
| 817 | return self |
| 818 | else: |
| 819 | # Rename |
| 820 | return type(self)( |
| 821 | key, |
| 822 | self.func, |
| 823 | *self.args, |
| 824 | **self.kwargs, |
| 825 | **extra_kwargs, |
| 826 | ) |
| 827 | |
| 828 | |
| 829 | class NestedContainer(Task, Iterable): |