Run ruff check --fix

awscli/customizations/cloudformation/artifact_exporter.py

@@ -94,7 +94,7 @@ def parse_s3_url(url,
             return result
 
     raise ValueError("URL given to the parse method is not a valid S3 url "
-                     "{0}".format(url))
+                     f"{url}")
 
 
 def upload_local_artifacts(resource_id, resource_dict, property_name,
@@ -134,8 +134,8 @@ def upload_local_artifacts(resource_id, resource_dict, property_name,
         # This check is supporting the case where your resource does not
         # refer to local artifacts
         # Nothing to do if property value is an S3 URL
-        LOG.debug("Property {0} of {1} is already a S3 URL"
-                  .format(property_name, resource_id))
+        LOG.debug(f"Property {property_name} of {resource_id} is already a S3 URL"
+                  )
         return local_path
 
     local_path = make_abs_path(parent_dir, local_path)
@@ -181,7 +181,7 @@ def zip_folder(folder_path):
 
 
 def make_zip(filename, source_root):
-    zipfile_name = "{0}.zip".format(filename)
+    zipfile_name = f"{filename}.zip"
     source_root = os.path.abspath(source_root)
     with open(zipfile_name, 'wb') as f:
         zip_file = zipfile.ZipFile(f, 'w', zipfile.ZIP_DEFLATED)
@@ -216,7 +216,7 @@ def copy_to_temp_dir(filepath):
     return tmp_dir
 
 
-class Resource(object):
+class Resource:
     """
     Base class representing a CloudFormation resource that can be exported
     """
@@ -241,8 +241,8 @@ def export(self, resource_id, resource_dict, parent_dir):
             return
 
         if isinstance(property_value, dict):
-            LOG.debug("Property {0} of {1} resource is not a URL"
-                      .format(self.PROPERTY_NAME, resource_id))
+            LOG.debug(f"Property {self.PROPERTY_NAME} of {resource_id} resource is not a URL"
+                      )
             return
 
         # If property is a file but not a zip file, place file in temp
@@ -576,7 +576,7 @@ def include_transform_export_handler(template_dict, uploader, parent_dir):
 }
 
 
-class Template(object):
+class Template:
     """
     Class to export a CloudFormation template
     """
@@ -590,8 +590,8 @@ def __init__(self, template_path, parent_dir, uploader,
 
         if not (is_local_folder(parent_dir) and os.path.isabs(parent_dir)):
             raise ValueError("parent_dir parameter must be "
-                             "an absolute path to a folder {0}"
-                             .format(parent_dir))
+                             f"an absolute path to a folder {parent_dir}"
+                             )
 
         abs_template_path = make_abs_path(parent_dir, template_path)
         template_dir = os.path.dirname(abs_template_path)

awscli/customizations/cloudformation/deployer.py

@@ -28,7 +28,7 @@
                 "ChangeSetResult", ["changeset_id", "changeset_type"])
 
 
-class Deployer(object):
+class Deployer:
 
     def __init__(self, cloudformation_client,
                  changeset_prefix="awscli-cloudformation-package-deploy-"):
@@ -62,9 +62,8 @@ def has_stack(self, stack_name):
             # the exception msg to understand the nature of this exception.
             msg = str(e)
 
-            if "Stack with id {0} does not exist".format(stack_name) in msg:
-                LOG.debug("Stack with id {0} does not exist".format(
-                    stack_name))
+            if f"Stack with id {stack_name} does not exist" in msg:
+                LOG.debug(f"Stack with id {stack_name} does not exist")
                 return False
             else:
                 # We don't know anything about this exception. Don't handle
@@ -86,7 +85,7 @@ def create_changeset(self, stack_name, cfn_template,
         """
 
         now = datetime.utcnow().isoformat()
-        description = "Created by AWS CLI at {0} UTC".format(now)
+        description = f"Created by AWS CLI at {now} UTC"
 
         # Each changeset will get a unique name based on time
         changeset_name = self.changeset_prefix + str(int(time.time()))
@@ -173,9 +172,9 @@ def wait_for_changeset(self, changeset_id, stack_name):
                             "No updates are to be performed" in reason:
                     raise exceptions.ChangeEmptyError(stack_name=stack_name)
 
-            raise RuntimeError("Failed to create the changeset: {0} "
-                               "Status: {1}. Reason: {2}"
-                               .format(ex, status, reason))
+            raise RuntimeError(f"Failed to create the changeset: {ex} "
+                               f"Status: {status}. Reason: {reason}"
+                               )
 
     def execute_changeset(self, changeset_id, stack_name,
                           disable_rollback=False):
@@ -203,8 +202,8 @@ def wait_for_execute(self, stack_name, changeset_type):
         elif changeset_type == "UPDATE":
             waiter = self._client.get_waiter("stack_update_complete")
         else:
-            raise RuntimeError("Invalid changeset type {0}"
-                               .format(changeset_type))
+            raise RuntimeError(f"Invalid changeset type {changeset_type}"
+                               )
 
         # Poll every 30 seconds. Polling too frequently risks hitting rate limits
         # on CloudFormation's DescribeStacks API

awscli/customizations/cloudformation/yamlhelper.py

@@ -73,7 +73,7 @@ def _add_yaml_1_1_boolean_resolvers(resolver_cls):
         '|true|True|TRUE|false|False|FALSE'
         '|on|On|ON|off|Off|OFF)$'
     )
-    boolean_first_chars = list(u'yYnNtTfFoO')
+    boolean_first_chars = list('yYnNtTfFoO')
     resolver_cls.add_implicit_resolver(
         'tag:yaml.org,2002:bool', boolean_regex, boolean_first_chars)
 

awscli/customizations/cloudtrail/validation.py

@@ -159,7 +159,7 @@ def create_digest_traverser(cloudtrail_client, organization_client,
         public_key_provider=PublicKeyProvider(cloudtrail_client))
 
 
-class S3ClientProvider(object):
+class S3ClientProvider:
     """Creates Amazon S3 clients and determines the region name of a client.
 
     This class will cache the location constraints of previously requested
@@ -215,7 +215,7 @@ def __init__(self, bucket, key):
         super(InvalidDigestFormat, self).__init__(message)
 
 
-class DigestProvider(object):
+class DigestProvider:
     """
     Retrieves digest keys and digests from Amazon S3.
 
@@ -348,7 +348,7 @@ def _create_digest_key_regex(self, key_prefix):
         return '^' + key + '$'
 
 
-class DigestTraverser(object):
+class DigestTraverser:
     """Retrieves and validates digests within a date range."""
     # These keys are required to be present before validating the contents
     # of a digest.
@@ -529,7 +529,7 @@ def _load_public_keys(self, start_date, end_date):
         return public_keys
 
 
-class Sha256RSADigestValidator(object):
+class Sha256RSADigestValidator:
     """
     Validates SHA256withRSA signed digests.
 
@@ -790,8 +790,8 @@ def _download_log(self, log):
                 self._on_log_invalid(log)
             else:
                 self._valid_logs += 1
-                self._write_status(('Log file\ts3://%s/%s\tvalid'
-                                    % (log['s3Bucket'], log['s3Object'])))
+                self._write_status('Log file\ts3://%s/%s\tvalid'
+                                    % (log['s3Bucket'], log['s3Object']))
         except ClientError as e:
             if e.response['Error']['Code'] != 'NoSuchKey':
                 raise

awscli/customizations/codeartifact/login.py

@@ -82,12 +82,11 @@ def _write_success_message(self, tool):
             self.expiration, datetime.now(tzutc())) + relativedelta(seconds=30)
         expiration_message = get_relative_expiration_time(remaining)
 
-        sys.stdout.write('Successfully configured {} to use '
-                         'AWS CodeArtifact repository {} '
-                         .format(tool, self.repository_endpoint))
+        sys.stdout.write(f'Successfully configured {tool} to use '
+                         f'AWS CodeArtifact repository {self.repository_endpoint} '
+                         )
         sys.stdout.write(os.linesep)
-        sys.stdout.write('Login expires in {} at {}'.format(
-            expiration_message, self.expiration))
+        sys.stdout.write(f'Login expires in {expiration_message} at {self.expiration}')
         sys.stdout.write(os.linesep)
 
     def _run_commands(self, tool, commands, dry_run=False):
@@ -126,7 +125,7 @@ def get_commands(cls, endpoint, auth_token, **kwargs):
 class SwiftLogin(BaseLogin):
 
     DEFAULT_NETRC_FMT = \
-        u'machine {hostname} login token password {auth_token}'
+        'machine {hostname} login token password {auth_token}'
 
     NETRC_REGEX_FMT = \
         r'(?P<entry_start>\bmachine\s+{escaped_hostname}\s+login\s+\S+\s+password\s+)' \
@@ -176,7 +175,7 @@ def _update_netrc_entry(self, hostname, new_entry, netrc_path):
         if not os.path.isfile(netrc_path):
             self._create_netrc_file(netrc_path, new_entry)
         else:
-            with open(netrc_path, 'r') as f:
+            with open(netrc_path) as f:
                 contents = f.read()
             escaped_auth_token = self.auth_token.replace('\\', r'\\')
             new_contents = re.sub(
@@ -356,7 +355,7 @@ def _get_source_to_url_dict(self):
         return source_to_url_dict
 
     def _get_source_name(self, codeartifact_url, source_dict):
-        default_name = '{}/{}'.format(self.domain, self.repository)
+        default_name = f'{self.domain}/{self.repository}'
 
         # Check if the CodeArtifact URL is already present in the
         # NuGet.Config file. If the URL already exists, use the source name
@@ -465,7 +464,7 @@ def get_scope(cls, namespace):
         if namespace.startswith('@'):
             scope = namespace
         else:
-            scope = '@{}'.format(namespace)
+            scope = f'@{namespace}'
 
         if not valid_scope_name.match(scope):
             raise ValueError(
@@ -481,7 +480,7 @@ def get_commands(cls, endpoint, auth_token, **kwargs):
         scope = kwargs.get('scope')
 
         # prepend scope if it exists
-        registry = '{}:registry'.format(scope) if scope else 'registry'
+        registry = f'{scope}:registry' if scope else 'registry'
 
         # set up the codeartifact repository as the npm registry.
         commands.append(
@@ -491,17 +490,13 @@ def get_commands(cls, endpoint, auth_token, **kwargs):
         repo_uri = urlsplit(endpoint)
 
         # configure npm to always require auth for the repository.
-        always_auth_config = '//{}{}:always-auth'.format(
-            repo_uri.netloc, repo_uri.path
-        )
+        always_auth_config = f'//{repo_uri.netloc}{repo_uri.path}:always-auth'
         commands.append(
             [cls.NPM_CMD, 'config', 'set', always_auth_config, 'true']
         )
 
         # set auth info for the repository.
-        auth_token_config = '//{}{}:_authToken'.format(
-            repo_uri.netloc, repo_uri.path
-        )
+        auth_token_config = f'//{repo_uri.netloc}{repo_uri.path}:_authToken'
         commands.append(
             [cls.NPM_CMD, 'config', 'set', auth_token_config, auth_token]
         )
@@ -749,7 +744,7 @@ def _get_namespace(self, tool, parsed_args):
 
         if not namespace_compatible and parsed_args.namespace:
             raise ValueError(
-                'Argument --namespace is not supported for {}'.format(tool)
+                f'Argument --namespace is not supported for {tool}'
             )
         else:
             return parsed_args.namespace

awscli/customizations/configservice/subscribe.py

@@ -130,7 +130,7 @@ def _setup_clients(self, parsed_globals):
                                                           **client_args)
 
 
-class S3BucketHelper(object):
+class S3BucketHelper:
     def __init__(self, s3_client):
         self._s3_client = s3_client
 
@@ -158,7 +158,7 @@ def _create_bucket(self, bucket):
         self._s3_client.create_bucket(**params)
 
 
-class SNSTopicHelper(object):
+class SNSTopicHelper:
     def __init__(self, sns_client):
         self._sns_client = sns_client
 

awscli/customizations/configure/__init__.py

@@ -18,7 +18,7 @@
 _WHITESPACE = ' \t'
 
 
-class ConfigValue(object):
+class ConfigValue:
 
     def __init__(self, value, config_type, config_variable):
         self.value = value

awscli/customizations/configure/configure.py

@@ -40,7 +40,7 @@ def register_configure_cmd(cli):
                  ConfigureCommand.add_command)
 
 
-class InteractivePrompter(object):
+class InteractivePrompter:
 
     def get_value(self, current_value, config_name, prompt_text=''):
         if config_name in ('aws_access_key_id', 'aws_secret_access_key'):

awscli/customizations/configure/exportcreds.py

@@ -46,7 +46,7 @@ def convert_botocore_credentials(credentials):
     )
 
 
-class BaseCredentialFormatter(object):
+class BaseCredentialFormatter:
 
     FORMAT = None
     DOCUMENTATION = ""

awscli/customizations/configure/get.py

@@ -53,7 +53,7 @@ def _run_main(self, args, parsed_globals):
         else:
             value = self._get_dotted_config_value(varname)
 
-        LOG.debug(u'Config value retrieved: %s' % value)
+        LOG.debug('Config value retrieved: %s' % value)
 
         if isinstance(value, str):
             self._stream.write(value)

awscli/customizations/configure/importer.py

@@ -100,7 +100,7 @@ class CredentialParserError(Exception):
     pass
 
 
-class CSVCredentialParser(object):
+class CSVCredentialParser:
     _USERNAME_HEADER = 'User Name'
     _AKID_HEADER = 'Access Key ID'
     _SAK_HEADER = 'Secret Access key'
@@ -185,7 +185,7 @@ def parse_credentials(self, contents):
         return self._convert_rows_to_credentials(parsed_rows)
 
 
-class CredentialImporter(object):
+class CredentialImporter:
     def __init__(self, writer):
         self._config_writer = writer
 

awscli/customizations/configure/sso.py

@@ -105,7 +105,7 @@ def _is_comma_separated_list(self, value):
         return True
 
 
-class PTKPrompt(object):
+class PTKPrompt:
     _DEFAULT_PROMPT_FORMAT = '{prompt_text} [{current_value}]: '
 
     def __init__(self, prompter=None):
@@ -424,7 +424,7 @@ def _prompt_for_account(self, sso, sso_token):
             sso_account_id = self._handle_single_account(accounts)
         else:
             sso_account_id = self._handle_multiple_accounts(accounts)
-        uni_print('Using the account ID {}\n'.format(sso_account_id))
+        uni_print(f'Using the account ID {sso_account_id}\n')
         self._new_profile_config_values['sso_account_id'] = sso_account_id
         return sso_account_id
 
@@ -458,7 +458,7 @@ def _prompt_for_role(self, sso, sso_token, sso_account_id):
             sso_role_name = self._handle_single_role(roles)
         else:
             sso_role_name = self._handle_multiple_roles(roles)
-        uni_print('Using the role name "{}"\n'.format(sso_role_name))
+        uni_print(f'Using the role name "{sso_role_name}"\n')
         self._new_profile_config_values['sso_role_name'] = sso_role_name
         return sso_role_name
 
@@ -634,7 +634,7 @@ def _prompt_for_sso_account_and_role(self, sso, sso_token):
             sso_account_id = self._prompt_for_account(sso, sso_token)
             sso_role_name = self._prompt_for_role(
                 sso, sso_token, sso_account_id)
-        except sso.exceptions.UnauthorizedException as e:
+        except sso.exceptions.UnauthorizedException:
             uni_print(
                 'Unable to list AWS accounts and/or roles. '
                 'Skipping configuring AWS credential provider for profile.\n'

awscli/customizations/configure/writer.py

@@ -16,7 +16,7 @@
 from . import SectionNotFoundError
 
 
-class ConfigFileWriter(object):
+class ConfigFileWriter:
     SECTION_REGEX = re.compile(r'^\s*\[(?P<header>[^]]+)\]')
     OPTION_REGEX = re.compile(
         r'(?P<option>[^:=][^:=]*)'
@@ -56,7 +56,7 @@ def update_config(self, new_values, config_filename):
             self._create_file(config_filename)
             self._write_new_section(section_name, new_values, config_filename)
             return
-        with open(config_filename, 'r') as f:
+        with open(config_filename) as f:
             contents = f.readlines()
         # We can only update a single section at a time so we first need
         # to find the section in question