@provide_session
    def get_task_instances(self, state=None, session=None):
        """
        Returns the task instances for this dag run
        """