return settings.ALLOW_FUTURE_EXEC_DATES and not self.timetable.can_run
    @provide_session
    def get_concurrency_reached(self, session=NEW_SESSION) -> bool:
        """
        Returns a boolean indicating whether the max_active_tasks limit for this DAG
        has been reached
        """
        TI = TaskInstance
        qry = session.query(func.count(TI.task_id)).filter(
            TI.dag_id == self.dag_id,
            TI.state == State.RUNNING,
        )
        return qry.scalar() >= self.max_active_tasks
    @property
    def concurrency_reached(self):
        """This attribute is deprecated. Please use `airflow.models.DAG.get_concurrency_reached` method."""
        warnings.warn(
            "This attribute is deprecated. Please use `airflow.models.DAG.get_concurrency_reached` method.",
            RemovedInAirflow3Warning,
            stacklevel=2,
        )
        return self.get_concurrency_reached()
    @provide_session
    def get_is_active(self, session=NEW_SESSION) -> None:
        """Returns a boolean indicating whether this DAG is active"""
        return session.query(DagModel.is_active).filter(DagModel.dag_id == self.dag_id).scalar()
    @provide_session
    def get_is_paused(self, session=NEW_SESSION) -> None: