### Description

A count of datasets that are currently registered/declared in an Airflow deployment.

### Use case/motivation

Would be nice to see how deployments are adopting datasets.

### Related issues

_No response_

### Are you willing to submit a PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

"""
        Find all dags that were not updated by Dag Processor recently and mark them as inactive.
        In case one of DagProcessors is stopped (in case there are multiple of them
        for different dag folders), it's dags are never marked as inactive.
        Also remove dags from SerializedDag table.
        Executed on schedule only if [scheduler]standalone_dag_processor is True.
        """
        self.log.debug("Checking dags not parsed within last %s seconds.", self._dag_stale_not_seen_duration)
        limit_lpt = timezone.utcnow() - timedelta(seconds=self._dag_stale_not_seen_duration)
        stale_dags = (
            session.query(DagModel).filter(DagModel.is_active, DagModel.last_parsed_time < limit_lpt).all()
        )
        if not stale_dags:
            self.log.debug("Not stale dags found.")
            return
        self.log.info("Found (%d) stales dags not parsed after %s.", len(stale_dags), limit_lpt)
        for dag in stale_dags:
            dag.is_active = False
            SerializedDagModel.remove_dag(dag_id=dag.dag_id, session=session)
        session.flush()
    @provide_session
    def _orphan_unreferenced_datasets(self, session: Session = NEW_SESSION) -> None: