{CI} Fix all style errors for pylint 4.x compatibility (#33347)

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Author: YangAn-microsoft

src/azure-cli-core/azure/cli/core/util.py

@@ -576,11 +576,10 @@ def get_file_yaml(file_path, throw_on_empty=True):
 
 
 def read_file_content(file_path, allow_binary=False):
-    from codecs import open as codecs_open
     # Note, always put 'utf-8-sig' first, so that BOM in WinOS won't cause trouble.
     for encoding in ['utf-8-sig', 'utf-8', 'utf-16', 'utf-16le', 'utf-16be']:
         try:
-            with codecs_open(file_path, encoding=encoding) as f:
+            with open(file_path, mode='r', encoding=encoding) as f:
                 logger.debug("attempting to read file %s as %s", file_path, encoding)
                 return f.read()
         except (UnicodeError, UnicodeDecodeError):

src/azure-cli/azure/cli/command_modules/apim/custom.py

@@ -27,7 +27,7 @@
 from azure.cli.core.util import sdk_no_wait
 from azure.cli.core.util import get_logger
 from azure.cli.core.azclierror import (RequiredArgumentMissingError, MutuallyExclusiveArgumentError,
-                                       InvalidArgumentValueError)
+                                       InvalidArgumentValueError, CLIError)
 from azure.mgmt.apimanagement.models import (ApiManagementServiceResource, ApiManagementServiceIdentity,
                                              ApiManagementServiceSkuProperties,
                                              ApiManagementServiceBackupRestoreParameters,
@@ -555,11 +555,13 @@ def apim_api_export(client, resource_group_name, service_name, api_id, export_fo
 
     # Obtain link from the response
     response_dict = api_export_result_to_dict(response)
+    link = None
     try:
         # Extract the link from the response where results are stored
         link = response_dict['additional_properties']['properties']['value']['link']
     except KeyError:
         logger.warning("Error exporting api from APIManagement. The expected link is not present in the response.")
+        raise CLIError("Failed to export API: link not found in response")
 
     # Determine the file extension based on the mappedFormat
     if mappedFormat in ['swagger-link', 'openapi+json-link']:
@@ -577,12 +579,15 @@ def apim_api_export(client, resource_group_name, service_name, api_id, export_fo
     full_path = os.path.join(file_path, file_name)
 
     # Get the results from the link where the API Export Results are stored
+    exportedResults = None
     try:
         exportedResults = requests.get(link, timeout=30)
         if not exportedResults.ok:
-            logger.warning("Got bad status from APIManagement during API Export:%s, {exportedResults.status_code}")
+            logger.warning("Got bad status from APIManagement during API Export: %s", exportedResults.status_code)
+            raise CLIError(f"Failed to export API: Got status code {exportedResults.status_code}")
     except requests.exceptions.ReadTimeout:
         logger.warning("Timed out while exporting api from APIManagement.")
+        raise CLIError("Failed to export API: Request timed out")
 
     try:
         # Try to parse as JSON

src/azure-cli/azure/cli/command_modules/appservice/custom.py

@@ -66,7 +66,6 @@
                     retryable_method,
                     raise_missing_token_suggestion,
                     _get_location_from_resource_group,
-                    _list_app,
                     is_functionapp,
                     is_linux_webapp,
                     _rename_server_farm_props,

src/azure-cli/azure/cli/command_modules/batch/_command_type.py

@@ -756,10 +756,11 @@ def get_track1_rest_names(self, cls):
     def _resolve_track1_type_hint(self, type_hint):
         """Resolve type hints to the legacy track1 type string format."""
         args = get_args(type_hint)
+        none_type = None.__class__
 
         # Optional[T] / Union[..., None] -> select the best non-None candidate.
-        if type(None) in args:
-            non_none_args = [arg for arg in args if arg is not type(None)]
+        if none_type in args:
+            non_none_args = [arg for arg in args if arg is not none_type]
             preferred_args = [arg for arg in non_none_args if arg != str] or non_none_args
             selected = preferred_args[0] if preferred_args else type_hint
             return self.convert_to_track1_type(str(selected))

src/azure-cli/azure/cli/command_modules/botservice/custom.py

@@ -59,7 +59,7 @@ def __prepare_configuration_file(cmd, resource_group_name, kudu_client, folder_p
                                if setting['name'] not in ignorable_settings}
         existing = None
         if not os.path.exists(app_settings_path):
-            logger.info('App settings not found at %s, defaulting app settings to {}.', app_settings_path)
+            logger.info('App settings not found at %s, defaulting app settings to empty.', app_settings_path)
             existing = {}
         else:
             with open(app_settings_path, 'r') as f:
@@ -271,7 +271,10 @@ def download_app(cmd, client, resource_group_name, resource_name, file_save_path
     if (os.path.exists(os.path.join(folder_path, 'PostDeployScripts', 'deploy.cmd.template')) and
             os.path.exists(os.path.join(folder_path, 'deploy.cmd'))):
 
-        logger.info('Post deployment scripts and deploy.cmd found in source under folder %s. Copying deploy.cmd.')
+        logger.info(
+            'Post deployment scripts and deploy.cmd found in source under folder %s. Copying deploy.cmd.',
+            folder_path
+        )
 
         shutil.copyfile(os.path.join(folder_path, 'deploy.cmd'),
                         os.path.join(folder_path, 'PostDeployScripts', 'deploy.cmd.template'))
@@ -318,8 +321,8 @@ def download_app(cmd, client, resource_group_name, resource_name, file_save_path
             existing = None
             if not os.path.exists(app_settings_path):
 
-                logger.info('App settings not found at %s, defaulting app settings to {}.', app_settings_path)
-                existing = '{}'
+                logger.info('App settings not found at %s, defaulting app settings to empty.', app_settings_path)
+                existing = {}
             else:
                 with open(app_settings_path, 'r') as f:
                     existing = json.load(f)

src/azure-cli/azure/cli/command_modules/containerapp/_utils.py

@@ -895,23 +895,6 @@ def _remove_env_vars(existing_env_vars, remove_env_vars):
             logger.warning("Environment variable {} does not exist.".format(old_env_var))  # pylint: disable=logging-format-interpolation
 
 
-def _remove_env_vars(existing_env_vars, remove_env_vars):
-    for old_env_var in remove_env_vars:
-
-        # Check if updating existing env var
-        is_existing = False
-        for index, value in enumerate(existing_env_vars):
-            existing_env_var = value
-            if existing_env_var["name"].lower() == old_env_var.lower():
-                is_existing = True
-                existing_env_vars.pop(index)
-                break
-
-        # If not updating existing env var, add it as a new env var
-        if not is_existing:
-            logger.warning("Environment variable {} does not exist.".format(old_env_var))  # pylint: disable=logging-format-interpolation
-
-
 def _add_or_update_tags(containerapp_def, tags):
     if 'tags' not in containerapp_def:
         if tags:

src/azure-cli/azure/cli/command_modules/cosmosdb/custom.py

@@ -2784,13 +2784,6 @@ def cli_cosmosdb_managed_cassandra_datacenter_update(client,
     return client.begin_create_update(resource_group_name, cluster_name, data_center_name, data_center_resource)
 
 
-def _handle_exists_exception(http_response_error):
-
-    if http_response_error.status_code == 404:
-        return False
-    raise http_response_error
-
-
 def process_restorable_databases(restorable_databases, database_name):
 
     latest_database_delete_time = datetime.datetime.utcfromtimestamp(0)

src/azure-cli/azure/cli/command_modules/synapse/manual/operations/sqlpoolblobauditingpolicy.py

@@ -861,41 +861,6 @@ def _audit_policy_validate_arguments(
         raise CLIError('event-hub-authorization-rule-id must be specified if event-hub-target-state is enabled')
 
 
-def _get_diagnostic_settings_url(
-        cmd,
-        resource_group_name,
-        workspace_name,
-        sql_pool_name=None):
-
-    from azure.cli.core.commands.client_factory import get_subscription_id
-
-    diag_settings = '/subscriptions/{}/resourceGroups/{}/providers/Microsoft.Synapse/workspaces/{}'.format(
-        get_subscription_id(cmd.cli_ctx),
-        resource_group_name, workspace_name)
-
-    if sql_pool_name is not None:
-        diag_settings = diag_settings + '/sqlpools/{}'.format(sql_pool_name)
-
-    return diag_settings
-
-
-def _get_diagnostic_settings(
-        cmd,
-        resource_group_name,
-        workspace_name,
-        sql_pool_name=None):
-    '''
-    Common code to get server or database diagnostic settings
-    '''
-
-    diagnostic_settings_url = _get_diagnostic_settings_url(
-        cmd=cmd, resource_group_name=resource_group_name,
-        workspace_name=workspace_name, sql_pool_name=sql_pool_name)
-    azure_monitor_client = cf_monitor(cmd.cli_ctx)
-
-    return list(azure_monitor_client.diagnostic_settings.list(diagnostic_settings_url))
-
-
 def workspace_audit_policy_show(
         cmd,
         client,