'Returns a dag run for the given run date, which will be matched to an existing
dag run if available or create a new dag run otherwise. If the max_active_runs
limit is reached, this function will return None.
:param run_date: the execution date for the dag run
:type run_date: datetime
:param session: the database session object
:type session: Session
:return: a DagRun in state RUNNING or None'

@provide_session
def _get_dag_run(self, run_date, session=None):

    run_id = BackfillJob.ID_FORMAT_PREFIX.format(run_date.isoformat())
    respect_dag_max_active_limit = (True if (self.dag.schedule_interval and (not self.dag.is_subdag)) else False)
    current_active_dag_count = self.dag.get_num_active_runs(external_trigger=False)
    run = DagRun.find(dag_id=self.dag.dag_id, execution_date=run_date, session=session)
    if ((run is not None) and (len(run) > 0)):
        run = run[0]
        if (run.state == State.RUNNING):
            respect_dag_max_active_limit = False
    else:
        run = None
    if (respect_dag_max_active_limit and (current_active_dag_count >= self.dag.max_active_runs)):
        return None
    run = (run or self.dag.create_dagrun(run_id=run_id, execution_date=run_date, start_date=datetime.now(), state=State.RUNNING, external_trigger=False, session=session))
    run.dag = self.dag
    run.state = State.RUNNING
    run.run_id = run_id
    run.verify_integrity(session=session)
    return run
