airflow/models/dag.py
def partial_subset(
        self,
        task_ids_or_regex: Union[str, RePatternType, Iterable[str]],
        include_downstream=False,
        include_upstream=True,
        include_direct_upstream=False,
    ):
        """
        Returns a subset of the current dag as a deep copy of the current dag
        based on a regex that should match one or many tasks, and includes
        upstream and downstream neighbours based on the flag passed.

        :param task_ids_or_regex: Either a list of task_ids, or a regex to
            match against task ids (as a string, or compiled regex pattern).
        :type task_ids_or_regex: [str] or str or re.Pattern
        :param include_downstream: Include all downstream tasks of matched
            tasks, in addition to matched tasks.
        :param include_upstream: Include all upstream tasks of matched tasks,
            in addition to matched tasks.
        """
        # deep-copying self.task_dict and self._task_group takes a long time, and we don't want all
        # the tasks anyway, so we copy the tasks manually later
        memo = {id(self.task_dict): None, id(self._task_group): None}
        dag = copy.deepcopy(self, memo)  # type: ignore

        if isinstance(task_ids_or_regex, (str, RePatternType)):
            matched_tasks = [t for t in self.tasks if re.findall(task_ids_or_regex, t.task_id)]
        else:
            matched_tasks = [t for t in self.tasks if t.task_id in task_ids_or_regex]

        also_include = []
        for t in matched_tasks:
            if include_downstream:
                also_include += t.get_flat_relatives(upstream=False)
            if include_upstream:
                also_include += t.get_flat_relatives(upstream=True)
            elif include_direct_upstream:
                also_include += t.upstream_list

        # Compiling the unique list of tasks that made the cut
        # Make sure to not recursively deepcopy the dag while copying the task
        dag.task_dict = {
            t.task_id: copy.deepcopy(t, {id(t.dag): dag})  # type: ignore
            for t in matched_tasks + also_include
        }

        def filter_task_group(group, parent_group):
            """Exclude tasks not included in the subdag from the given TaskGroup."""
            copied = copy.copy(group)
            copied.used_group_ids = set(copied.used_group_ids)
            copied._parent_group = parent_group

            copied.children = {}

            for child in group.children.values():
                if isinstance(child, BaseOperator):
                    if child.task_id in dag.task_dict:
                        copied.children[child.task_id] = dag.task_dict[child.task_id]
                    else:
                        copied.used_group_ids.discard(child.task_id)
                else:
                    filtered_child = filter_task_group(child, copied)

                    # Only include this child TaskGroup if it is non-empty.
                    if filtered_child.children:
                        copied.children[child.group_id] = filtered_child

            return copied

        dag._task_group = filter_task_group(self._task_group, None)

        # Removing upstream/downstream references to tasks and TaskGroups that did not make
        # the cut.
        subdag_task_groups = dag.task_group.get_task_group_dict()
        for group in subdag_task_groups.values():
            group.upstream_group_ids = group.upstream_group_ids.intersection(subdag_task_groups.keys())
            group.downstream_group_ids = group.downstream_group_ids.intersection(subdag_task_groups.keys())
            group.upstream_task_ids = group.upstream_task_ids.intersection(dag.task_dict.keys())
            group.downstream_task_ids = group.downstream_task_ids.intersection(dag.task_dict.keys())

        for t in dag.tasks:
            # Removing upstream/downstream references to tasks that did not
            # make the cut
            t._upstream_task_ids = t.upstream_task_ids.intersection(dag.task_dict.keys())
            t._downstream_task_ids = t.downstream_task_ids.intersection(dag.task_dict.keys())

        if len(dag.tasks) < len(self.tasks):
            dag.partial = True

        return dag

        Returns a subset of the current dag as a deep copy of the current dag
        based on a regex that should match one or many tasks, and includes
        upstream and downstream neighbours based on the flag passed.

        :param task_ids_or_regex: Either a list of task_ids, or a regex to
            match against task ids (as a string, or compiled regex pattern).
        :type task_ids_or_regex: [str] or str or re.Pattern
        :param include_downstream: Include all downstream tasks of matched
            tasks, in addition to matched tasks.
        :param include_upstream: Include all upstream tasks of matched tasks,
            in addition to matched tasks.
        
Returns a subset of the current dag as a deep copy of the current dag
based on a regex that should match one or many tasks, and includes
upstream and downstream neighbours based on the flag passed.
Returns a subset of the current dag as a deep copy of the current dag based on a regex that should match one or many tasks and includes upstream and downstream neighbours based on the flag passed .
def partial_subset(
        self,
        task_ids_or_regex: Union[str, RePatternType, Iterable[str]],
        include_downstream=False,
        include_upstream=True,
        include_direct_upstream=False,
    ):
        memo = {id(self.task_dict): None, id(self._task_group): None}
        dag = copy.deepcopy(self, memo)  
        if isinstance(task_ids_or_regex, (str, RePatternType)):
            matched_tasks = [t for t in self.tasks if re.findall(task_ids_or_regex, t.task_id)]
        else:
            matched_tasks = [t for t in self.tasks if t.task_id in task_ids_or_regex]
        also_include = []
        for t in matched_tasks:
            if include_downstream:
                also_include += t.get_flat_relatives(upstream=False)
            if include_upstream:
                also_include += t.get_flat_relatives(upstream=True)
            elif include_direct_upstream:
                also_include += t.upstream_list
        dag.task_dict = {
            t.task_id: copy.deepcopy(t, {id(t.dag): dag})  
            for t in matched_tasks + also_include
        }
        def filter_task_group(group, parent_group):
            copied = copy.copy(group)
            copied.used_group_ids = set(copied.used_group_ids)
            copied._parent_group = parent_group
            copied.children = {}
            for child in group.children.values():
                if isinstance(child, BaseOperator):
                    if child.task_id in dag.task_dict:
                        copied.children[child.task_id] = dag.task_dict[child.task_id]
                    else:
                        copied.used_group_ids.discard(child.task_id)
                else:
                    filtered_child = filter_task_group(child, copied)
                    if filtered_child.children:
                        copied.children[child.group_id] = filtered_child
            return copied
        dag._task_group = filter_task_group(self._task_group, None)
        subdag_task_groups = dag.task_group.get_task_group_dict()
        for group in subdag_task_groups.values():
            group.upstream_group_ids = group.upstream_group_ids.intersection(subdag_task_groups.keys())
            group.downstream_group_ids = group.downstream_group_ids.intersection(subdag_task_groups.keys())
            group.upstream_task_ids = group.upstream_task_ids.intersection(dag.task_dict.keys())
            group.downstream_task_ids = group.downstream_task_ids.intersection(dag.task_dict.keys())
        for t in dag.tasks:
            t._upstream_task_ids = t.upstream_task_ids.intersection(dag.task_dict.keys())
            t._downstream_task_ids = t.downstream_task_ids.intersection(dag.task_dict.keys())
        if len(dag.tasks) < len(self.tasks):
            dag.partial = True
        return dag
def partial_subset ( self , task_ids_or_regex : Union [ str , RePatternType , Iterable [ str ] ] , include_downstream = False , include_upstream = True , include_direct_upstream = False , ) : memo = { id ( self . task_dict ) : None , id ( self . _task_group ) : None } dag = copy . deepcopy ( self , memo ) if isinstance ( task_ids_or_regex , ( str , RePatternType ) ) : matched_tasks = [ t for t in self . tasks if re . findall ( task_ids_or_regex , t . task_id ) ] else : matched_tasks = [ t for t in self . tasks if t . task_id in task_ids_or_regex ] also_include = [ ] for t in matched_tasks : if include_downstream : also_include += t . get_flat_relatives ( upstream = False ) if include_upstream : also_include += t . get_flat_relatives ( upstream = True ) elif include_direct_upstream : also_include += t . upstream_list dag . task_dict = { t . task_id : copy . deepcopy ( t , { id ( t . dag ) : dag } ) for t in matched_tasks + also_include } def filter_task_group ( group , parent_group ) : """Exclude tasks not included in the subdag from the given TaskGroup.""" copied = copy . copy ( group ) copied . used_group_ids = set ( copied . used_group_ids ) copied . _parent_group = parent_group copied . children = { } for child in group . children . values ( ) : if isinstance ( child , BaseOperator ) : if child . task_id in dag . task_dict : copied . children [ child . task_id ] = dag . task_dict [ child . task_id ] else : copied . used_group_ids . discard ( child . task_id ) else : filtered_child = filter_task_group ( child , copied ) if filtered_child . children : copied . children [ child . group_id ] = filtered_child return copied dag . _task_group = filter_task_group ( self . _task_group , None ) subdag_task_groups = dag . task_group . get_task_group_dict ( ) for group in subdag_task_groups . values ( ) : group . upstream_group_ids = group . upstream_group_ids . intersection ( subdag_task_groups . keys ( ) ) group . downstream_group_ids = group . downstream_group_ids . intersection ( subdag_task_groups . keys ( ) ) group . upstream_task_ids = group . upstream_task_ids . intersection ( dag . task_dict . keys ( ) ) group . downstream_task_ids = group . downstream_task_ids . intersection ( dag . task_dict . keys ( ) ) for t in dag . tasks : t . _upstream_task_ids = t . upstream_task_ids . intersection ( dag . task_dict . keys ( ) ) t . _downstream_task_ids = t . downstream_task_ids . intersection ( dag . task_dict . keys ( ) ) if len ( dag . tasks ) < len ( self . tasks ) : dag . partial = True return dag
Returns a subset of the current dag as a deep copy of the current dag
based on a regex that should match one or many tasks, and includes
upstream and downstream neighbours based on the flag passed.
Returns a subset of the current dag as a deep copy of the current dag based on a regex that should match one or many tasks and includes upstream and downstream neighbours based on the flag passed .
"""
        Returns a subset of the current dag as a deep copy of the current dag
        based on a regex that should match one or many tasks, and includes
        upstream and downstream neighbours based on the flag passed.

        :param task_ids_or_regex: Either a list of task_ids, or a regex to
            match against task ids (as a string, or compiled regex pattern).
        :type task_ids_or_regex: [str] or str or re.Pattern
        :param include_downstream: Include all downstream tasks of matched
            tasks, in addition to matched tasks.
        :param include_upstream: Include all upstream tasks of matched tasks,
            in addition to matched tasks.
        """ # deep-copying self.task_dict and self._task_group takes a long time, and we don't want all # the tasks anyway, so we copy the tasks manually later # type: ignore # Compiling the unique list of tasks that made the cut # Make sure to not recursively deepcopy the dag while copying the task # type: ignore """Exclude tasks not included in the subdag from the given TaskGroup.""" # Only include this child TaskGroup if it is non-empty. # Removing upstream/downstream references to tasks and TaskGroups that did not make # the cut. # Removing upstream/downstream references to tasks that did not # make the cut
self
task_ids_or_regex
Union[str, RePatternType, Iterable[str]]
Either a list of task_ids, or a regex to
match against task ids (as a string, or compiled regex pattern).
Either a list of task_ids or a regex to match against task ids ( as a string or compiled regex pattern ) .
include_downstream
Include all downstream tasks of matched
tasks, in addition to matched tasks.
Include all downstream tasks of matched tasks in addition to matched tasks .
include_upstream
Include all upstream tasks of matched tasks,
in addition to matched tasks.
Include all upstream tasks of matched tasks in addition to matched tasks .
include_direct_upstream