@provide_session<EOL><INDENT>def verify_integrity(self, session=None):<DEDENT>
from airflow.models.taskinstance import TaskInstance  <EOL>dag = self.get_dag()<EOL>tis = self.get_task_instances(session=session)<EOL>task_ids = []<EOL>for ti in tis:<EOL><INDENT>task_ids.append(ti.task_id)<EOL>task = None<EOL>try:<EOL><INDENT>task = dag.get_task(ti.task_id)<EOL><DEDENT>except AirflowException:<EOL><INDENT>if ti.state == State.REMOVED:<EOL><INDENT>pass  <EOL><DEDENT>elif self.state is not State.RUNNING and not dag.partial:<EOL><INDENT>self.log.warning("<STR_LIT>"<EOL>"<STR_LIT>".format(ti, dag))<EOL>Stats.incr(<EOL>"<STR_LIT>".format(dag.dag_id), <NUM_LIT:1>, <NUM_LIT:1>)<EOL>ti.state = State.REMOVED<EOL><DEDENT><DEDENT>is_task_in_dag = task is not None<EOL>should_restore_task = is_task_in_dag and ti.state == State.REMOVED<EOL>if should_restore_task:<EOL><INDENT>self.log.info("<STR_LIT>"<EOL>"<STR_LIT>".format(ti, dag))<EOL>Stats.incr("<STR_LIT>".format(dag.dag_id), <NUM_LIT:1>, <NUM_LIT:1>)<EOL>ti.state = State.NONE<EOL><DEDENT><DEDENT>for task in six.itervalues(dag.task_dict):<EOL><INDENT>if task.start_date > self.execution_date and not self.is_backfill:<EOL><INDENT>continue<EOL><DEDENT>if task.task_id not in task_ids:<EOL><INDENT>Stats.incr(<EOL>"<STR_LIT>".format(task.__class__.__name__),<EOL><NUM_LIT:1>, <NUM_LIT:1>)<EOL>ti = TaskInstance(task, self.execution_date)<EOL>session.add(ti)<EOL><DEDENT><DEDENT>session.commit()<EOL>
Verifies the DagRun by checking for removed tasks or tasks that are not in the
database yet. It will set state to removed or add the task if required.