'Returns the number of active "running" dag runs
:param external_trigger: True for externally triggered active dag runs
:type external_trigger: bool
:param session:
:return: number greater than 0 for active dag runs'

@provide_session
def get_num_active_runs(self, external_trigger=None, session=None):

    query = session.query(DagRun).filter((DagRun.dag_id == self.dag_id)).filter((DagRun.state == State.RUNNING))
    if (external_trigger is not None):
        query = query.filter((DagRun.external_trigger == external_trigger))
    return query.count()
