https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
tis = tis.where(TI.state == state)
            else:
                
                if None in state:
                    if all(x is None for x in state):
                        tis = tis.where(TI.state.is_(None))
                    else:
                        not_none_state = (s for s in state if s)
                        tis = tis.where(or_(TI.state.in_(not_none_state), TI.state.is_(None)))
                else:
                    tis = tis.where(TI.state.in_(state))
        if dag and dag.partial:
            tis = tis.where(TI.task_id.in_(dag.task_ids))
        return session.scalars(tis).all()
    @provide_session
    def get_task_instances(
        self,
        state: Iterable[TaskInstanceState | None] | None = None,
        session: Session = NEW_SESSION,
    ) -> list[TI]:
        """
        Returns the task instances for this dag run.
        Redirect to DagRun.fetch_task_instances method.
        Keep this method because it is widely used across the code.
        """
        return DagRun.fetch_task_instances(
            dag_id=self.dag_id, run_id=self.run_id, dag=self.dag, state=state, session=session
        )
    @provide_session
    def get_task_instance(