Repository: apache/airflow
File: airflow-core/src/airflow/models/taskinstance.py (Python)

Code before:
        """
        Are all conditions met for this task instance to be run given the context for the dependencies.

        (e.g. a task instance being force run from the UI will ignore some dependencies).

        :param dep_context: The execution context that determines the dependencies that should be evaluated.
        :param session: database session
        :param verbose: whether log details on failed dependencies on info or debug log level
        """
        dep_context = dep_context or DepContext()
        if self.state == TaskInstanceState.UP_FOR_RESCHEDULE:
            # This DepContext is used when a task instance is in UP_FOR_RESCHEDULE state.
            #
            # Tasks can be put into UP_FOR_RESCHEDULE by the task runner itself (e.g. when
            # the worker cannot load the Dag or task). In this case, the scheduler must respect
            # the task instance's reschedule_date before scheduling it again.
            #
            # ReadyToRescheduleDep is the only dependency that enforces this time-based gating.
            # We therefore extend the normal scheduling dependency set with it, instead of
            # modifying the global scheduler dependencies.
            dep_context = DepContext(
                deps=dep_context.deps | {ReadyToRescheduleDep()},
                flag_upstream_failed=dep_context.flag_upstream_failed,
                ignore_unmapped_tasks=dep_context.ignore_unmapped_tasks,
                finished_tis=dep_context.finished_tis,
            )
        failed = False
        verbose_aware_logger = self.log.info if verbose else self.log.debug
        for dep_status in self.get_failed_dep_statuses(dep_context=dep_context, session=session):
            failed = True

            verbose_aware_logger(
                "Dependencies not met for %s, dependency '%s' FAILED: %s",
                self,
                dep_status.dep_name,
                dep_status.reason,
            )

        if failed:
            return False

        verbose_aware_logger("Dependencies all met for dep_context=%s ti=%s", dep_context.description, self)
        return True

    @provide_session
    def get_failed_dep_statuses(self, dep_context: DepContext | None = None, session: Session = NEW_SESSION):
        """Get failed Dependencies."""
        if TYPE_CHECKING:
            assert self.task is not None
        dep_context = dep_context or DepContext()
        for dep in dep_context.deps | self.task.deps:

Review comment (suggestion):
Can this not modify `dep_context` in-place?

```python
dep_context.deps.add(ReadyToRescheduleDep())
```

Code after:
    ) -> bool:
        """
        Are all conditions met for this task instance to be run given the context for the dependencies.

        (e.g. a task instance being force run from the UI will ignore some dependencies).

        :param dep_context: The execution context that determines the dependencies that should be evaluated.
        :param session: database session
        :param verbose: whether log details on failed dependencies on info or debug log level
        """
        dep_context = dep_context or DepContext()
        if self.state == TaskInstanceState.UP_FOR_RESCHEDULE:
            # This DepContext is used when a task instance is in UP_FOR_RESCHEDULE state.
            #
            # Tasks can be put into UP_FOR_RESCHEDULE by the task runner itself (e.g. when
            # the worker cannot load the Dag or task). In this case, the scheduler must respect
            # the task instance's reschedule_date before scheduling it again.
            #
            # ReadyToRescheduleDep is the only dependency that enforces this time-based gating.
            # We therefore extend the normal scheduling dependency set with it, instead of
            # modifying the global scheduler dependencies.
            dep_context.deps.add(ReadyToRescheduleDep())
        failed = False
        verbose_aware_logger = self.log.info if verbose else self.log.debug
        for dep_status in self.get_failed_dep_statuses(dep_context=dep_context, session=session):
            failed = True

            verbose_aware_logger(
                "Dependencies not met for %s, dependency '%s' FAILED: %s",
                self,
                dep_status.dep_name,
                dep_status.reason,
            )

        if failed:
            return False

        verbose_aware_logger("Dependencies all met for dep_context=%s ti=%s", dep_context.description, self)
        return True

    @provide_session
    def get_failed_dep_statuses(self, dep_context: DepContext | None = None, session: Session = NEW_SESSION):
        """Get failed Dependencies."""
        if TYPE_CHECKING:
            assert self.task is not None
        dep_context = dep_context or DepContext()
        for dep in dep_context.deps | self.task.deps:
            for dep_status in dep.get_dep_statuses(self, session, dep_context):
                self.log.debug(
                    "%s dependency '%s' PASSED: %s, %s",
                    self,