'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.'

@provide_session
def verify_integrity(self, session=None):

    dag = self.get_dag()
    tis = self.get_task_instances(session=session)
    task_ids = []
    for ti in tis:
        task_ids.append(ti.task_id)
        try:
            dag.get_task(ti.task_id)
        except AirflowException:
            if ((self.state is not State.RUNNING) and (not dag.partial)):
                ti.state = State.REMOVED
    for task in dag.tasks:
        if task.adhoc:
            continue
        if (task.task_id not in task_ids):
            ti = TaskInstance(task, self.execution_date)
            session.add(ti)
    session.commit()
