Implement BatchDefinition::get_batch_identifiers_list

great_expectations/core/batch_definition.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, Generic, Optional, TypeVar
+from typing import TYPE_CHECKING, Any, Dict, Generic, List, Optional, TypeVar
 
 from great_expectations._docs_decorators import public_api
 from great_expectations.compatibility import pydantic
@@ -81,6 +81,24 @@ def get_batch(self, batch_parameters: Optional[BatchParameters] = None) -> Batch
         batch_request = self.build_batch_request(batch_parameters=batch_parameters)
         return self.data_asset.get_batch(batch_request)
 
+    @public_api
+    def get_batch_identifiers_list(
+        self, batch_parameters: Optional[BatchParameters] = None
+    ) -> List[Dict]:
+        """
+        Retrieves a list of available batch identifiers.
+        These identifiers can be used to fetch specific batches via batch_options.
+
+        Args:
+            batch_parameters: Additional parameters to be used in fetching the batch identifiers
+            list.
+
+        Returns:
+            A list of batch identifiers.
+        """
+        batch_request = self.build_batch_request(batch_parameters=batch_parameters)
+        return self.data_asset.get_batch_identifiers_list(batch_request)
+
     def is_fresh(self) -> BatchDefinitionFreshnessDiagnostics:
         diagnostics = self._is_added()
         if not diagnostics.success:

tests/core/test_batch_definition.py

@@ -97,6 +97,47 @@ def test_get_batch_retrieves_only_batch(mocker: pytest_mock.MockFixture):
     mock_asset.get_batch.assert_called_once_with(batch_definition.build_batch_request())
 
 
+@pytest.mark.unit
+def test_get_batch_identifiers_list(mocker: pytest_mock.MockFixture):
+    # Arrange
+    batch_definition = BatchDefinition[None](name="test_batch_definition")
+    mock_asset = mocker.Mock(spec=DataAsset)
+    batch_definition.set_data_asset(mock_asset)
+
+    mock_batch_identifiers_list = [{"foo": "bar"}, {"baz": "qux"}]
+    mock_asset.get_batch_identifiers_list.return_value = mock_batch_identifiers_list
+
+    # Act
+    batch_identifiers_list = batch_definition.get_batch_identifiers_list()
+
+    # Assert
+    assert batch_identifiers_list == mock_batch_identifiers_list
+    mock_asset.get_batch_identifiers_list.assert_called_once_with(
+        batch_definition.build_batch_request()
+    )
+
+
+@pytest.mark.unit
+def test_get_batch_identifiers_list_with_batch_parameters(mocker: pytest_mock.MockFixture):
+    # Arrange
+    batch_definition = BatchDefinition[None](name="test_batch_definition")
+    mock_asset = mocker.Mock(spec=DataAsset)
+    batch_definition.set_data_asset(mock_asset)
+
+    mock_batch_identifiers_list = [{"foo": "bar"}, {"baz": "qux"}]
+    mock_asset.get_batch_identifiers_list.return_value = mock_batch_identifiers_list
+
+    # Act
+    batch_parameters: BatchParameters = {"path": "my_path"}
+    batch_identifiers_list = batch_definition.get_batch_identifiers_list(batch_parameters)
+
+    # Assert
+    assert batch_identifiers_list == mock_batch_identifiers_list
+    mock_asset.get_batch_identifiers_list.assert_called_once_with(
+        batch_definition.build_batch_request(batch_parameters)
+    )
+
+
 @pytest.mark.unit
 def test_identifier_bundle_success(in_memory_runtime_context):
     context = in_memory_runtime_context