Improve user experience in multi clouds scenario

.chronus/changes/auto-microsoft-multi-clouds-2025-1-12-10-23-18.md

@@ -0,0 +1,8 @@
+---
+changeKind: feature
+packages:
+  - "@autorest/python"
+  - "@azure-tools/typespec-python"
+---
+
+Improve user experience in multi clouds scenario

packages/autorest.python/autorest/m2r.py

@@ -3,8 +3,7 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 # --------------------------------------------------------------------------
-"""An autorest MD to RST plugin.
-"""
+"""An autorest MD to RST plugin."""
 import logging
 from typing import Any, Dict, Set, Union
 

packages/autorest.python/autorest/m4reformatter/__init__.py

@@ -4,8 +4,7 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 # --------------------------------------------------------------------------
-"""The modelerfour reformatter autorest plugin.
-"""
+"""The modelerfour reformatter autorest plugin."""
 import re
 import copy
 import logging

packages/autorest.python/autorest/multiapi/models/imports.py

@@ -123,9 +123,14 @@ def _add_import(
             ],
             convert_list_to_tuple(name_import),
         )
-        self._imports.setdefault(typing_section, {}).setdefault(import_type, {}).setdefault(from_section, set()).add(
-            name_input
+        target_values = (
+            self._imports.setdefault(typing_section, {}).setdefault(import_type, {}).setdefault(from_section, set())
         )
+        if isinstance(target_values, list):
+            if name_input not in target_values:
+                target_values.append(name_input)
+        else:
+            target_values.add(name_input)
 
     def add_submodule_import(
         self,

packages/autorest.python/autorest/multiapi/templates/multiapi_service_client.py.jinja2

@@ -18,6 +18,7 @@ def __init__(
 {% set async_prefix = "Async" if async_mode else "" %}
 {% set a_prefix = "a" if async_mode else "" %}
 {% set await = "await " if async_mode else "" %}
+{% set credential_scopes = "credential_scopes=credential_scopes, " if code_model.config.credential_scopes is not none and code_model.azure_arm else "" %}
 # coding=utf-8
 # --------------------------------------------------------------------------
 # Copyright (c) Microsoft Corporation. All rights reserved.
@@ -77,6 +78,7 @@ class {{ code_model.client.name }}({% if code_model.operation_mixin_group.mixin_
     )
 
     {{ method_signature()|indent }}
+        {% if not code_model.azure_arm %}
         {% if not code_model.client.host_value %}
         {% for parameterized_host_template, api_versions in code_model.client.parameterized_host_template_to_api_version|dictsort %}
         {% set if_statement = "if" if loop.first else "elif" %}
@@ -86,17 +88,26 @@ class {{ code_model.client.name }}({% if code_model.operation_mixin_group.mixin_
         else:
             raise ValueError("API version {} is not available".format(api_version))
         {% endif %}
+        {% endif %}
         if api_version:
             kwargs.setdefault('api_version', api_version)
-        self._config = {{ code_model.client.name }}Configuration({{ code_model.global_parameters.call }}{{ ", " if code_model.global_parameters.call }}**kwargs)
+        {% if credential_scopes %}
+        _cloud = kwargs.pop("cloud_setting", None) or settings.current.azure_cloud  # type: ignore
+        _endpoints = get_arm_endpoints(_cloud)
+        if not base_url:
+            base_url = _endpoints["resource_manager"]
+        credential_scopes = kwargs.pop("credential_scopes", _endpoints["credential_scopes"])
+        {% endif %}
+        self._config = {{ code_model.client.name }}Configuration({{ code_model.global_parameters.call }}{{ ", " if code_model.global_parameters.call }}{{ credential_scopes }}**kwargs)
         _policies = kwargs.pop("policies", None)
         if _policies is None:
             _policies = [
                 {% for p in config_policies %}
                 {{ p }},
                 {% endfor %}
             ]
-        self._client: {{ async_prefix }}{{ code_model.client.pipeline_client }} = {{ async_prefix }}{{ code_model.client.pipeline_client }}(base_url={{ code_model.host_variable_name }}, policies=_policies, **kwargs)
+        {% set host_variable_name = "cast(str, " + code_model.host_variable_name + ")" if credential_scopes else code_model.host_variable_name %}
+        self._client: {{ async_prefix }}{{ code_model.client.pipeline_client }} = {{ async_prefix }}{{ code_model.client.pipeline_client }}(base_url={{ host_variable_name }}, policies=_policies, **kwargs)
         super({{ code_model.client.name }}, self).__init__(
             api_version=api_version,
             profile=profile

packages/autorest.python/autorest/preprocess.py

@@ -3,8 +3,7 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 # --------------------------------------------------------------------------
-"""The preprocessing autorest plugin.
-"""
+"""The preprocessing autorest plugin."""
 from typing import Dict, Any
 from pygen.preprocess import PreProcessPlugin
 from . import YamlUpdatePluginAutorest

packages/autorest.python/samples/specification/azure-mgmt-test/test/azure-mgmt-test/azure/mgmt/test/_auto_rest_swagger_bat_array_service.py

@@ -7,13 +7,15 @@
 # --------------------------------------------------------------------------
 
 from copy import deepcopy
-from typing import Any, TYPE_CHECKING
+from typing import Any, Optional, TYPE_CHECKING, cast
 from typing_extensions import Self
 
 from azure.core.pipeline import policies
 from azure.core.rest import HttpRequest, HttpResponse
+from azure.core.settings import settings
 from azure.mgmt.core import ARMPipelineClient
 from azure.mgmt.core.policies import ARMAutoResourceProviderRegistrationPolicy
+from azure.mgmt.core.tools import get_arm_endpoints
 
 from . import models as _models
 from ._configuration import AutoRestSwaggerBATArrayServiceConfiguration
@@ -31,12 +33,20 @@ class AutoRestSwaggerBATArrayService:  # pylint: disable=client-accepts-api-vers
     :vartype array: azure.mgmt.test.operations.ArrayOperations
     :param credential: Credential needed for the client to connect to Azure. Required.
     :type credential: ~azure.core.credentials.TokenCredential
-    :param base_url: Service URL. Default value is "http://localhost:3000".
+    :param base_url: Service URL. Default value is None.
     :type base_url: str
     """
 
-    def __init__(self, credential: "TokenCredential", base_url: str = "http://localhost:3000", **kwargs: Any) -> None:
-        self._config = AutoRestSwaggerBATArrayServiceConfiguration(credential=credential, **kwargs)
+    def __init__(self, credential: "TokenCredential", base_url: Optional[str] = None, **kwargs: Any) -> None:
+        _cloud = kwargs.pop("cloud_setting", None) or settings.current.azure_cloud  # type: ignore
+        _endpoints = get_arm_endpoints(_cloud)
+        if not base_url:
+            base_url = _endpoints["resource_manager"]
+        credential_scopes = kwargs.pop("credential_scopes", _endpoints["credential_scopes"])
+        self._config = AutoRestSwaggerBATArrayServiceConfiguration(
+            credential=credential, credential_scopes=credential_scopes, **kwargs
+        )
+
         _policies = kwargs.pop("policies", None)
         if _policies is None:
             _policies = [
@@ -55,7 +65,7 @@ def __init__(self, credential: "TokenCredential", base_url: str = "http://localh
                 policies.SensitiveHeaderCleanupPolicy(**kwargs) if self._config.redirect_policy else None,
                 self._config.http_logging_policy,
             ]
-        self._client: ARMPipelineClient = ARMPipelineClient(base_url=base_url, policies=_policies, **kwargs)
+        self._client: ARMPipelineClient = ARMPipelineClient(base_url=cast(str, base_url), policies=_policies, **kwargs)
 
         client_models = {k: v for k, v in _models.__dict__.items() if isinstance(v, type)}
         self._serialize = Serializer(client_models)

packages/autorest.python/samples/specification/azure-mgmt-test/test/azure-mgmt-test/azure/mgmt/test/aio/_auto_rest_swagger_bat_array_service.py

@@ -7,13 +7,15 @@
 # --------------------------------------------------------------------------
 
 from copy import deepcopy
-from typing import Any, Awaitable, TYPE_CHECKING
+from typing import Any, Awaitable, Optional, TYPE_CHECKING, cast
 from typing_extensions import Self
 
 from azure.core.pipeline import policies
 from azure.core.rest import AsyncHttpResponse, HttpRequest
+from azure.core.settings import settings
 from azure.mgmt.core import AsyncARMPipelineClient
 from azure.mgmt.core.policies import AsyncARMAutoResourceProviderRegistrationPolicy
+from azure.mgmt.core.tools import get_arm_endpoints
 
 from .. import models as _models
 from .._serialization import Deserializer, Serializer
@@ -31,14 +33,20 @@ class AutoRestSwaggerBATArrayService:  # pylint: disable=client-accepts-api-vers
     :vartype array: azure.mgmt.test.aio.operations.ArrayOperations
     :param credential: Credential needed for the client to connect to Azure. Required.
     :type credential: ~azure.core.credentials_async.AsyncTokenCredential
-    :param base_url: Service URL. Default value is "http://localhost:3000".
+    :param base_url: Service URL. Default value is None.
     :type base_url: str
     """
 
-    def __init__(
-        self, credential: "AsyncTokenCredential", base_url: str = "http://localhost:3000", **kwargs: Any
-    ) -> None:
-        self._config = AutoRestSwaggerBATArrayServiceConfiguration(credential=credential, **kwargs)
+    def __init__(self, credential: "AsyncTokenCredential", base_url: Optional[str] = None, **kwargs: Any) -> None:
+        _cloud = kwargs.pop("cloud_setting", None) or settings.current.azure_cloud  # type: ignore
+        _endpoints = get_arm_endpoints(_cloud)
+        if not base_url:
+            base_url = _endpoints["resource_manager"]
+        credential_scopes = kwargs.pop("credential_scopes", _endpoints["credential_scopes"])
+        self._config = AutoRestSwaggerBATArrayServiceConfiguration(
+            credential=credential, credential_scopes=credential_scopes, **kwargs
+        )
+
         _policies = kwargs.pop("policies", None)
         if _policies is None:
             _policies = [
@@ -57,7 +65,9 @@ def __init__(
                 policies.SensitiveHeaderCleanupPolicy(**kwargs) if self._config.redirect_policy else None,
                 self._config.http_logging_policy,
             ]
-        self._client: AsyncARMPipelineClient = AsyncARMPipelineClient(base_url=base_url, policies=_policies, **kwargs)
+        self._client: AsyncARMPipelineClient = AsyncARMPipelineClient(
+            base_url=cast(str, base_url), policies=_policies, **kwargs
+        )
 
         client_models = {k: v for k, v in _models.__dict__.items() if isinstance(v, type)}
         self._serialize = Serializer(client_models)

packages/autorest.python/samples/specification/azure-test/test/azure-test/azure/test/_generated/_client.py

@@ -33,6 +33,7 @@ def __init__(  # pylint: disable=missing-client-constructor-parameter-credential
         self, *, endpoint: str = "http://localhost:3000", **kwargs: Any
     ) -> None:
         self._config = AutoRestSwaggerBATArrayServiceConfiguration(**kwargs)
+
         _policies = kwargs.pop("policies", None)
         if _policies is None:
             _policies = [

packages/autorest.python/samples/specification/azure-test/test/azure-test/azure/test/_generated/aio/_client.py

@@ -33,6 +33,7 @@ def __init__(  # pylint: disable=missing-client-constructor-parameter-credential
         self, *, endpoint: str = "http://localhost:3000", **kwargs: Any
     ) -> None:
         self._config = AutoRestSwaggerBATArrayServiceConfiguration(**kwargs)
+
         _policies = kwargs.pop("policies", None)
         if _policies is None:
             _policies = [

packages/autorest.python/samples/specification/azure_key_credential/generated/azure/key/credential/sample/_client.py

@@ -35,6 +35,7 @@ def __init__(
         self, credential: AzureKeyCredential, *, endpoint: str = "http://localhost:3000", **kwargs: Any
     ) -> None:
         self._config = AutoRestHeadTestServiceConfiguration(credential=credential, **kwargs)
+
         _policies = kwargs.pop("policies", None)
         if _policies is None:
             _policies = [

packages/autorest.python/samples/specification/azure_key_credential/generated/azure/key/credential/sample/aio/_client.py

@@ -35,6 +35,7 @@ def __init__(
         self, credential: AzureKeyCredential, *, endpoint: str = "http://localhost:3000", **kwargs: Any
     ) -> None:
         self._config = AutoRestHeadTestServiceConfiguration(credential=credential, **kwargs)
+
         _policies = kwargs.pop("policies", None)
         if _policies is None:
             _policies = [

packages/autorest.python/samples/specification/basic/generated/azure/basic/sample/_client.py

@@ -32,6 +32,7 @@ def __init__(  # pylint: disable=missing-client-constructor-parameter-credential
         self, *, endpoint: str = "http://localhost:3000", **kwargs: Any
     ) -> None:
         self._config = AutoRestHeadTestServiceConfiguration(**kwargs)
+
         _policies = kwargs.pop("policies", None)
         if _policies is None:
             _policies = [

packages/autorest.python/samples/specification/basic/generated/azure/basic/sample/aio/_client.py

@@ -32,6 +32,7 @@ def __init__(  # pylint: disable=missing-client-constructor-parameter-credential
         self, *, endpoint: str = "http://localhost:3000", **kwargs: Any
     ) -> None:
         self._config = AutoRestHeadTestServiceConfiguration(**kwargs)
+
         _policies = kwargs.pop("policies", None)
         if _policies is None:
             _policies = [

packages/autorest.python/samples/specification/directives/generated/azure/directives/sample/_client.py

@@ -32,6 +32,7 @@ def __init__(  # pylint: disable=missing-client-constructor-parameter-credential
         self, *, endpoint: str = "http://localhost:3000", **kwargs: Any
     ) -> None:
         self._config = PollingPagingExampleConfiguration(**kwargs)
+
         _policies = kwargs.pop("policies", None)
         if _policies is None:
             _policies = [

packages/autorest.python/samples/specification/directives/generated/azure/directives/sample/aio/_client.py

@@ -32,6 +32,7 @@ def __init__(  # pylint: disable=missing-client-constructor-parameter-credential
         self, *, endpoint: str = "http://localhost:3000", **kwargs: Any
     ) -> None:
         self._config = PollingPagingExampleConfiguration(**kwargs)
+
         _policies = kwargs.pop("policies", None)
         if _policies is None:
             _policies = [

packages/autorest.python/samples/specification/management/generated/azure/mgmt/sample/_auto_rest_head_test_service.py

@@ -7,13 +7,15 @@
 # --------------------------------------------------------------------------
 
 from copy import deepcopy
-from typing import Any, Dict, TYPE_CHECKING
+from typing import Any, Dict, Optional, TYPE_CHECKING, cast
 from typing_extensions import Self
 
 from azure.core.pipeline import policies
 from azure.core.rest import HttpRequest, HttpResponse
+from azure.core.settings import settings
 from azure.mgmt.core import ARMPipelineClient
 from azure.mgmt.core.policies import ARMAutoResourceProviderRegistrationPolicy
+from azure.mgmt.core.tools import get_arm_endpoints
 
 from ._configuration import AutoRestHeadTestServiceConfiguration
 from ._serialization import Deserializer, Serializer
@@ -30,12 +32,20 @@ class AutoRestHeadTestService:  # pylint: disable=client-accepts-api-version-key
     :vartype http_success: azure.mgmt.sample.operations.HttpSuccessOperations
     :param credential: Credential needed for the client to connect to Azure. Required.
     :type credential: ~azure.core.credentials.TokenCredential
-    :param base_url: Service URL. Default value is "http://localhost:3000".
+    :param base_url: Service URL. Default value is None.
     :type base_url: str
     """
 
-    def __init__(self, credential: "TokenCredential", base_url: str = "http://localhost:3000", **kwargs: Any) -> None:
-        self._config = AutoRestHeadTestServiceConfiguration(credential=credential, **kwargs)
+    def __init__(self, credential: "TokenCredential", base_url: Optional[str] = None, **kwargs: Any) -> None:
+        _cloud = kwargs.pop("cloud_setting", None) or settings.current.azure_cloud  # type: ignore
+        _endpoints = get_arm_endpoints(_cloud)
+        if not base_url:
+            base_url = _endpoints["resource_manager"]
+        credential_scopes = kwargs.pop("credential_scopes", _endpoints["credential_scopes"])
+        self._config = AutoRestHeadTestServiceConfiguration(
+            credential=credential, credential_scopes=credential_scopes, **kwargs
+        )
+
         _policies = kwargs.pop("policies", None)
         if _policies is None:
             _policies = [
@@ -54,7 +64,7 @@ def __init__(self, credential: "TokenCredential", base_url: str = "http://localh
                 policies.SensitiveHeaderCleanupPolicy(**kwargs) if self._config.redirect_policy else None,
                 self._config.http_logging_policy,
             ]
-        self._client: ARMPipelineClient = ARMPipelineClient(base_url=base_url, policies=_policies, **kwargs)
+        self._client: ARMPipelineClient = ARMPipelineClient(base_url=cast(str, base_url), policies=_policies, **kwargs)
 
         client_models: Dict[str, Any] = {}
         self._serialize = Serializer(client_models)

packages/autorest.python/samples/specification/management/generated/azure/mgmt/sample/aio/_auto_rest_head_test_service.py

@@ -7,13 +7,15 @@
 # --------------------------------------------------------------------------
 
 from copy import deepcopy
-from typing import Any, Awaitable, Dict, TYPE_CHECKING
+from typing import Any, Awaitable, Dict, Optional, TYPE_CHECKING, cast
 from typing_extensions import Self
 
 from azure.core.pipeline import policies
 from azure.core.rest import AsyncHttpResponse, HttpRequest
+from azure.core.settings import settings
 from azure.mgmt.core import AsyncARMPipelineClient
 from azure.mgmt.core.policies import AsyncARMAutoResourceProviderRegistrationPolicy
+from azure.mgmt.core.tools import get_arm_endpoints
 
 from .._serialization import Deserializer, Serializer
 from ._configuration import AutoRestHeadTestServiceConfiguration
@@ -30,14 +32,20 @@ class AutoRestHeadTestService:  # pylint: disable=client-accepts-api-version-key
     :vartype http_success: azure.mgmt.sample.aio.operations.HttpSuccessOperations
     :param credential: Credential needed for the client to connect to Azure. Required.
     :type credential: ~azure.core.credentials_async.AsyncTokenCredential
-    :param base_url: Service URL. Default value is "http://localhost:3000".
+    :param base_url: Service URL. Default value is None.
     :type base_url: str
     """
 
-    def __init__(
-        self, credential: "AsyncTokenCredential", base_url: str = "http://localhost:3000", **kwargs: Any
-    ) -> None:
-        self._config = AutoRestHeadTestServiceConfiguration(credential=credential, **kwargs)
+    def __init__(self, credential: "AsyncTokenCredential", base_url: Optional[str] = None, **kwargs: Any) -> None:
+        _cloud = kwargs.pop("cloud_setting", None) or settings.current.azure_cloud  # type: ignore
+        _endpoints = get_arm_endpoints(_cloud)
+        if not base_url:
+            base_url = _endpoints["resource_manager"]
+        credential_scopes = kwargs.pop("credential_scopes", _endpoints["credential_scopes"])
+        self._config = AutoRestHeadTestServiceConfiguration(
+            credential=credential, credential_scopes=credential_scopes, **kwargs
+        )
+
         _policies = kwargs.pop("policies", None)
         if _policies is None:
             _policies = [
@@ -56,7 +64,9 @@ def __init__(
                 policies.SensitiveHeaderCleanupPolicy(**kwargs) if self._config.redirect_policy else None,
                 self._config.http_logging_policy,
             ]
-        self._client: AsyncARMPipelineClient = AsyncARMPipelineClient(base_url=base_url, policies=_policies, **kwargs)
+        self._client: AsyncARMPipelineClient = AsyncARMPipelineClient(
+            base_url=cast(str, base_url), policies=_policies, **kwargs
+        )
 
         client_models: Dict[str, Any] = {}
         self._serialize = Serializer(client_models)

packages/autorest.python/samples/specification/management/generated/setup.py

@@ -24,7 +24,7 @@
     include_package_data=True,
     install_requires=[
         "msrest>=0.7.1",
-        "azure-mgmt-core>=1.3.2",
+        "azure-mgmt-core>=1.5.0",
         "typing-extensions>=4.6.0",
     ],
     long_description="""\