9
9
from typing import Dict , List , Optional
10
10
11
11
from azure .ai .ml ._utils .utils import _get_mfe_url_override
12
- from azure .ai .ml .constants ._common import AZUREML_CLOUD_ENV_NAME
13
- from azure .ai .ml .constants ._common import ArmConstants
12
+ from azure .ai .ml .constants ._common import AZUREML_CLOUD_ENV_NAME , ArmConstants
14
13
from azure .core .rest import HttpRequest
15
14
from azure .mgmt .core import ARMPipelineClient
16
15
17
-
18
16
module_logger = logging .getLogger (__name__ )
19
17
20
18
@@ -72,7 +70,7 @@ def _get_cloud(cloud: str) -> Dict[str, str]:
72
70
arm_clouds = _get_clouds_by_metadata_url (arm_url )
73
71
try :
74
72
new_cloud = arm_clouds [cloud ]
75
- _environments .update (new_cloud )
73
+ _environments .update (new_cloud ) # type: ignore[arg-type]
76
74
return new_cloud
77
75
except KeyError as e :
78
76
raise Exception ('Unknown cloud environment "{0}".' .format (cloud )) from e
@@ -86,7 +84,7 @@ def _get_default_cloud_name() -> str:
86
84
return os .getenv (AZUREML_CLOUD_ENV_NAME , AzureEnvironments .ENV_DEFAULT )
87
85
88
86
89
- def _get_cloud_details (cloud : str = AzureEnvironments .ENV_DEFAULT ) -> Dict [str , str ]:
87
+ def _get_cloud_details (cloud : Optional [ str ] = AzureEnvironments .ENV_DEFAULT ) -> Dict [str , str ]:
90
88
"""Returns a Cloud endpoints object for the specified Azure Cloud.
91
89
92
90
:param cloud: cloud name
@@ -134,7 +132,7 @@ def _get_base_url_from_metadata(cloud_name: Optional[str] = None, is_local_mfe:
134
132
base_url = _get_mfe_url_override ()
135
133
if base_url is None :
136
134
cloud_details = _get_cloud_details (cloud_name )
137
- base_url = cloud_details .get (EndpointURLS .RESOURCE_MANAGER_ENDPOINT ).strip ("/" )
135
+ base_url = str ( cloud_details .get (EndpointURLS .RESOURCE_MANAGER_ENDPOINT ) ).strip ("/" )
138
136
return base_url
139
137
140
138
@@ -147,7 +145,7 @@ def _get_aml_resource_id_from_metadata(cloud_name: Optional[str] = None) -> str:
147
145
:rtype: str
148
146
"""
149
147
cloud_details = _get_cloud_details (cloud_name )
150
- aml_resource_id = cloud_details .get (EndpointURLS .AML_RESOURCE_ID ).strip ("/" )
148
+ aml_resource_id = str ( cloud_details .get (EndpointURLS .AML_RESOURCE_ID ) ).strip ("/" )
151
149
return aml_resource_id
152
150
153
151
@@ -160,7 +158,7 @@ def _get_active_directory_url_from_metadata(cloud_name: Optional[str] = None) ->
160
158
:rtype: str
161
159
"""
162
160
cloud_details = _get_cloud_details (cloud_name )
163
- active_directory_url = cloud_details .get (EndpointURLS .ACTIVE_DIRECTORY_ENDPOINT ).strip ("/" )
161
+ active_directory_url = str ( cloud_details .get (EndpointURLS .ACTIVE_DIRECTORY_ENDPOINT ) ).strip ("/" )
164
162
return active_directory_url
165
163
166
164
@@ -174,7 +172,7 @@ def _get_storage_endpoint_from_metadata(cloud_name: Optional[str] = None) -> str
174
172
"""
175
173
cloud_details = _get_cloud_details (cloud_name )
176
174
storage_endpoint = cloud_details .get (EndpointURLS .STORAGE_ENDPOINT )
177
- return storage_endpoint
175
+ return str ( storage_endpoint )
178
176
179
177
180
178
def _get_azure_portal_id_from_metadata (cloud_name : Optional [str ] = None ) -> str :
@@ -187,7 +185,7 @@ def _get_azure_portal_id_from_metadata(cloud_name: Optional[str] = None) -> str:
187
185
"""
188
186
cloud_details = _get_cloud_details (cloud_name )
189
187
azure_portal_id = cloud_details .get (EndpointURLS .AZURE_PORTAL_ENDPOINT )
190
- return azure_portal_id
188
+ return str ( azure_portal_id )
191
189
192
190
193
191
def _get_cloud_information_from_metadata (cloud_name : Optional [str ] = None , ** kwargs ) -> Dict :
@@ -199,10 +197,12 @@ def _get_cloud_information_from_metadata(cloud_name: Optional[str] = None, **kwa
199
197
:rtype: Dict
200
198
"""
201
199
cloud_details = _get_cloud_details (cloud_name )
202
- credential_scopes = _resource_to_scopes (cloud_details .get (EndpointURLS .RESOURCE_MANAGER_ENDPOINT ).strip ("/" ))
200
+ credential_scopes = _resource_to_scopes (
201
+ cloud_details .get (EndpointURLS .RESOURCE_MANAGER_ENDPOINT ).strip ("/" ) # type: ignore[union-attr]
202
+ )
203
203
204
204
# Update the kwargs with the cloud information
205
- client_kwargs = {"cloud" : cloud_name }
205
+ client_kwargs : Dict = {"cloud" : cloud_name }
206
206
if credential_scopes is not None :
207
207
client_kwargs ["credential_scopes" ] = credential_scopes
208
208
kwargs .update (client_kwargs )
@@ -219,7 +219,7 @@ def _get_registry_discovery_endpoint_from_metadata(cloud_name: Optional[str] = N
219
219
"""
220
220
cloud_details = _get_cloud_details (cloud_name )
221
221
registry_discovery_endpoint = cloud_details .get (EndpointURLS .REGISTRY_DISCOVERY_ENDPOINT )
222
- return registry_discovery_endpoint
222
+ return str ( registry_discovery_endpoint )
223
223
224
224
225
225
def _resource_to_scopes (resource : str ) -> List [str ]:
@@ -250,7 +250,7 @@ def _get_registry_discovery_url(cloud: dict, cloud_suffix: str = "") -> str:
250
250
"""
251
251
cloud_name = cloud ["name" ]
252
252
if cloud_name in _environments :
253
- return _environments [cloud_name ].registry_url
253
+ return _environments [cloud_name ].registry_url # type: ignore[attr-defined]
254
254
255
255
registry_discovery_region = os .environ .get (
256
256
ArmConstants .REGISTRY_DISCOVERY_REGION_ENV_NAME ,
0 commit comments