### 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)

num_finished_events = self._process_executor_events(session=session)
                if self.processor_agent:
                    self.processor_agent.heartbeat()
                
                self.heartbeat(only_if_necessary=True)
                
                next_event = timers.run(blocking=False)
                self.log.debug("Next timed event is in %f", next_event)
            self.log.debug("Ran scheduling loop in %.2f seconds", timer.duration)
            if not is_unit_test and not num_queued_tis and not num_finished_events:
                
                
                
                time.sleep(min(self._scheduler_idle_sleep_time, next_event if next_event else 0))
            if loop_count >= self.num_runs > 0:
                self.log.info(
                    "Exiting scheduler loop as requested number of runs (%d - got to %d) has been reached",
                    self.num_runs,
                    loop_count,
                )
                break
            if self.processor_agent and self.processor_agent.done:
                self.log.info(
                    "Exiting scheduler loop as requested DAG parse count (%d) has been reached after %d"
                    " scheduler loops",
                    self.num_times_parse_dags,
                    loop_count,
                )
                break
    def _do_scheduling(self, session: Session) -> int: