allow subscription-only requests for registry list operation

sdk/ml/azure-ai-ml/CHANGELOG.md

@@ -37,6 +37,7 @@
  - Added a `show_progress` parameter to MLClient for enable/disable progress bars of long running operations.
  - Support `month_days` in `RecurrencePattern` when using `RecurrenceSchedule`.
  - Support `ml_client.schedules.list` with `list_view_type`, default to `ENABLED_ONLY`.
+ - Registry list operation now accepts scope value to allow subscription-only based requests.
  - Add support for model sweeping and hyperparameter tuning in AutoML NLP jobs.
  - Added `ml_client.jobs.show_services()` operation.
 

sdk/ml/azure-ai-ml/azure/ai/ml/operations/_registry_operations.py

@@ -52,14 +52,19 @@ def __init__(
         self._init_kwargs = kwargs
 
     #@ monitor_with_activity(logger, "Registry.List", ActivityType.PUBLICAPI)
-    def list(self) -> Iterable[Registry]:
+    def list(self, *, scope: str = "resource_group") -> Iterable[Registry]:
         """List all registries that the user has access to in the current
-        resource group.
+        resource group or subscription.
 
+        :param scope: scope of the listing, "resource_group" or "subscription", defaults to "resource_group"
+        :type scope: str, optional
         :return: An iterator like instance of Registry objects
         :rtype: ~azure.core.paging.ItemPaged[Registry]
         """
-
+        if scope == "subscription":
+            return self._operation.list_by_subscription(
+                cls=lambda objs: [Registry._from_rest_object(obj) for obj in objs]
+            )
         return self._operation.list(cls=lambda objs: [Registry._from_rest_object(obj) for obj in objs], \
             resource_group_name=self._resource_group_name)
 

sdk/ml/azure-ai-ml/tests/registry/unittests/test_registry_operations.py

@@ -29,9 +29,18 @@ def mock_registry_operation(
 @pytest.mark.unittest
 class TestRegistryOperations:
     def test_list(self, mock_registry_operation: RegistryOperations) -> None:
+        # Test different input options for the scope value
         mock_registry_operation.list()
         mock_registry_operation._operation.list.assert_called_once()
 
+        mock_registry_operation.list(scope="invalid")
+        assert mock_registry_operation._operation.list.call_count == 2
+        mock_registry_operation._operation.list_by_subscription.assert_not_called()
+
+        mock_registry_operation.list(scope="subscription")
+        assert mock_registry_operation._operation.list.call_count == 2
+        mock_registry_operation._operation.list_by_subscription.assert_called_once()
+
     def test_get(self, mock_registry_operation: RegistryOperations, randstr: Callable[[], str]) -> None:
         mock_registry_operation.get(f"unittest_{randstr('reg_name')}")
         mock_registry_operation._operation.get.assert_called_once()