Used in https://github.com/mhenc/airflow/blob/master/airflow/dag_processing/manager.py#L613

should be straighforward
) -> Iterable[TaskInstance] | set[TaskInstanceKey]:
        TI = TaskInstance
        
        
        
        #
        
        
        result: set[TaskInstanceKey] = set()
        
        if as_pk_tuple:
            tis = session.query(TI.dag_id, TI.task_id, TI.run_id, TI.map_index)
        else:
            tis = session.query(TaskInstance)
        tis = tis.join(TaskInstance.dag_run)
        if include_subdags:
            
            conditions = []
            for dag in self.subdags + [self]:
                conditions.append(
                    (TaskInstance.dag_id == dag.dag_id) & TaskInstance.task_id.in_(dag.task_ids)
                )
            tis = tis.filter(or_(*conditions))
        elif self.partial:
            tis = tis.filter(TaskInstance.dag_id == self.dag_id, TaskInstance.task_id.in_(self.task_ids))
        else:
            tis = tis.filter(TaskInstance.dag_id == self.dag_id)
        if run_id:
            tis = tis.filter(TaskInstance.run_id == run_id)
        if start_date: