'This method checks whether a new DagRun needs to be created
for a DAG based on scheduling interval
Returns DagRun if one is scheduled. Otherwise returns None.'

@provide_session
def create_dag_run(self, dag, session=None):

    if dag.schedule_interval:
        active_runs = DagRun.find(dag_id=dag.dag_id, state=State.RUNNING, external_trigger=False, session=session)
        if ((len(active_runs) >= dag.max_active_runs) and (not dag.dagrun_timeout)):
            return
        timedout_runs = 0
        for dr in active_runs:
            if (dr.start_date and dag.dagrun_timeout and (dr.start_date < (datetime.now() - dag.dagrun_timeout))):
                dr.state = State.FAILED
                dr.end_date = datetime.now()
                timedout_runs += 1
        session.commit()
        if ((len(active_runs) - timedout_runs) >= dag.max_active_runs):
            return
        qry = session.query(func.max(DagRun.execution_date)).filter_by(dag_id=dag.dag_id).filter(or_((DagRun.external_trigger == False), DagRun.run_id.like((DagRun.ID_PREFIX + u'%'))))
        last_scheduled_run = qry.scalar()
        if ((dag.schedule_interval == u'@once') and last_scheduled_run):
            return None
        if (not dag.catchup):
            now = datetime.now()
            next_start = dag.following_schedule(now)
            last_start = dag.previous_schedule(now)
            if (next_start <= now):
                new_start = last_start
            else:
                new_start = dag.previous_schedule(last_start)
            if dag.start_date:
                if (new_start >= dag.start_date):
                    dag.start_date = new_start
            else:
                dag.start_date = new_start
        next_run_date = None
        if (not last_scheduled_run):
            task_start_dates = [t.start_date for t in dag.tasks]
            if task_start_dates:
                next_run_date = dag.normalize_schedule(min(task_start_dates))
                self.logger.debug(u'Next    run    date    based    on    tasks    {}'.format(next_run_date))
        else:
            next_run_date = dag.following_schedule(last_scheduled_run)
        last_run = dag.get_last_dagrun(session=session)
        if (last_run and next_run_date):
            while (next_run_date <= last_run.execution_date):
                next_run_date = dag.following_schedule(next_run_date)
        if dag.start_date:
            next_run_date = (dag.start_date if (not next_run_date) else max(next_run_date, dag.start_date))
            if (next_run_date == dag.start_date):
                next_run_date = dag.normalize_schedule(dag.start_date)
            self.logger.debug(u'Dag    start    date:    {}.    Next    run    date:    {}'.format(dag.start_date, next_run_date))
        if (next_run_date > datetime.now()):
            return
        if (dag.schedule_interval == u'@once'):
            period_end = next_run_date
        elif next_run_date:
            period_end = dag.following_schedule(next_run_date)
        if (next_run_date and dag.end_date and (next_run_date > dag.end_date)):
            return
        min_task_end_date = []
        task_end_dates = [t.end_date for t in dag.tasks if t.end_date]
        if task_end_dates:
            min_task_end_date = min(task_end_dates)
        if (next_run_date and min_task_end_date and (next_run_date > min_task_end_date)):
            return
        if (next_run_date and period_end and (period_end <= datetime.now())):
            next_run = dag.create_dagrun(run_id=(u'scheduled__' + next_run_date.isoformat()), execution_date=next_run_date, start_date=datetime.now(), state=State.RUNNING, external_trigger=False)
            return next_run
