https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
func.max(DatasetEvent.timestamp).label("lastUpdate"),
                    )
                    .join(
                        DagScheduleDatasetReference, DagScheduleDatasetReference.dataset_id == DatasetModel.id
                    )
                    .join(
                        DatasetDagRunQueue,
                        and_(
                            DatasetDagRunQueue.dataset_id == DatasetModel.id,
                            DatasetDagRunQueue.target_dag_id == DagScheduleDatasetReference.dag_id,
                        ),
                        isouter=True,
                    )
                    .join(
                        DatasetEvent,
                        DatasetEvent.dataset_id == DatasetModel.id,
                        isouter=True,
                    )
                    .where(DagScheduleDatasetReference.dag_id == dag_id, ~DatasetModel.is_orphaned)
                    .group_by(DatasetModel.id, DatasetModel.uri)
                    .order_by(DatasetModel.uri)
                )
            ]
        return (
            htmlsafe_json_dumps(data, separators=(",", ":"), dumps=flask.json.dumps),
            {"Content-Type": "application/json; charset=utf-8"},
        )
    @expose("/object/dataset_dependencies")
    @auth.has_access_dag("GET", DagAccessEntity.DEPENDENCIES)
    def dataset_dependencies(self):