'Returns the DagRun for this TaskInstance
:param session:
:return: DagRun'

@provide_session
def get_dagrun(self, session):

    dr = session.query(DagRun).filter((DagRun.dag_id == self.dag_id), (DagRun.execution_date == self.execution_date)).first()
    return dr
