feat: Add dataset_view parameter to get_dataset method

This commit introduces a new `dataset_view` parameter to the `get_dataset`
method in the BigQuery client. This allows you to specify the level of
detail (METADATA, ACL, FULL) returned when fetching a dataset.

The `DatasetView` enum has been added to `enums.py`.

Unit tests have been added to verify:
- Correct query parameter (`view`) formation for each enum value.
- Correct behavior when `dataset_view` is None.
- AttributeError is raised for invalid `dataset_view` types.

Author: google-labs-jules[bot]

google/cloud/bigquery/client.py

@@ -90,7 +90,7 @@
 from google.cloud.bigquery.dataset import Dataset
 from google.cloud.bigquery.dataset import DatasetListItem
 from google.cloud.bigquery.dataset import DatasetReference
-from google.cloud.bigquery.enums import AutoRowIDs
+from google.cloud.bigquery.enums import AutoRowIDs, DatasetView
 from google.cloud.bigquery.format_options import ParquetOptions
 from google.cloud.bigquery.job import (
     CopyJob,
@@ -849,6 +849,7 @@ def get_dataset(
         dataset_ref: Union[DatasetReference, str],
         retry: retries.Retry = DEFAULT_RETRY,
         timeout: TimeoutType = DEFAULT_TIMEOUT,
+        dataset_view: Optional[DatasetView] = None,
     ) -> Dataset:
         """Fetch the dataset referenced by ``dataset_ref``
 
@@ -866,6 +867,9 @@ def get_dataset(
             timeout (Optional[float]):
                 The number of seconds to wait for the underlying HTTP transport
                 before using ``retry``.
+            dataset_view (Optional[google.cloud.bigquery.enums.DatasetView]):
+                Specifies the level of detail to return. Defaults to None, which
+                returns the default representation of the dataset.
 
         Returns:
             google.cloud.bigquery.dataset.Dataset:
@@ -876,6 +880,9 @@ def get_dataset(
                 dataset_ref, default_project=self.project
             )
         path = dataset_ref.path
+        params: Dict[str, Any] = {}
+        if dataset_view is not None:
+            params["view"] = dataset_view.value
         span_attributes = {"path": path}
         api_response = self._call_api(
             retry,
@@ -884,6 +891,7 @@ def get_dataset(
             method="GET",
             path=path,
             timeout=timeout,
+            query_params=params if params else None,
         )
         return Dataset.from_api_repr(api_response)
 

google/cloud/bigquery/enums.py

@@ -80,6 +80,13 @@ class CreateDisposition(object):
     returned in the job result."""
 
 
+class DatasetView(enum.Enum):
+    DATASET_VIEW_UNSPECIFIED = "DATASET_VIEW_UNSPECIFIED"
+    METADATA = "METADATA"
+    ACL = "ACL"
+    FULL = "FULL"
+
+
 class DefaultPandasDTypes(enum.Enum):
     """Default Pandas DataFrem DTypes to convert BigQuery data. These
     Sentinel values are used instead of None to maintain backward compatibility,

tests/unit/test_client.py

@@ -61,6 +61,7 @@
 from google.cloud.bigquery import job as bqjob
 import google.cloud.bigquery._job_helpers
 from google.cloud.bigquery.dataset import DatasetReference
+from google.cloud.bigquery.enums import DatasetView
 from google.cloud.bigquery import exceptions
 from google.cloud.bigquery import ParquetOptions
 import google.cloud.bigquery.retry
@@ -807,6 +808,68 @@ def test_get_dataset(self):
 
         self.assertEqual(dataset.dataset_id, self.DS_ID)
 
+    @pytest.mark.parametrize(
+        "dataset_view_arg,expected_param_value",
+        [
+            (None, None),
+            (DatasetView.DATASET_VIEW_UNSPECIFIED, "DATASET_VIEW_UNSPECIFIED"),
+            (DatasetView.METADATA, "METADATA"),
+            (DatasetView.ACL, "ACL"),
+            (DatasetView.FULL, "FULL"),
+        ],
+    )
+    def test_get_dataset_with_dataset_view(
+        self, dataset_view_arg, expected_param_value
+    ):
+        path = "projects/%s/datasets/%s" % (self.PROJECT, self.DS_ID)
+        creds = _make_credentials()
+        http = object()
+        client = self._make_one(project=self.PROJECT, credentials=creds, _http=http)
+        resource = {
+            "id": "%s:%s" % (self.PROJECT, self.DS_ID),
+            "datasetReference": {"projectId": self.PROJECT, "datasetId": self.DS_ID},
+        }
+        conn = client._connection = make_connection(resource)
+        dataset_ref = DatasetReference(self.PROJECT, self.DS_ID)
+
+        dataset = client.get_dataset(dataset_ref, dataset_view=dataset_view_arg)
+
+        self.assertEqual(dataset.dataset_id, self.DS_ID)
+        expected_query_params = {}
+        if expected_param_value is not None:
+            expected_query_params["view"] = expected_param_value
+
+        conn.api_request.assert_called_once_with(
+            method="GET",
+            path="/%s" % path,
+            timeout=DEFAULT_TIMEOUT,
+            query_params=expected_query_params if expected_query_params else None,
+        )
+
+    @pytest.mark.parametrize(
+        "invalid_view_value",
+        [
+            "INVALID_STRING",
+            123,
+            123.45,
+            object(),
+        ],
+    )
+    def test_get_dataset_with_invalid_dataset_view(self, invalid_view_value):
+        path = "projects/%s/datasets/%s" % (self.PROJECT, self.DS_ID)
+        creds = _make_credentials()
+        http = object()
+        client = self._make_one(project=self.PROJECT, credentials=creds, _http=http)
+        resource = {
+            "id": "%s:%s" % (self.PROJECT, self.DS_ID),
+            "datasetReference": {"projectId": self.PROJECT, "datasetId": self.DS_ID},
+        }
+        conn = client._connection = make_connection(resource)
+        dataset_ref = DatasetReference(self.PROJECT, self.DS_ID)
+
+        with pytest.raises(AttributeError):
+            client.get_dataset(dataset_ref, dataset_view=invalid_view_value)
+
     def test_ensure_bqstorage_client_creating_new_instance(self):
         bigquery_storage = pytest.importorskip("google.cloud.bigquery_storage")