Python botocore.exceptions 模块，ProfileNotFound() 实例源码

def inject_assume_role_provider_cache(session, **kwargs):
    try:
        cred_chain = session.get_component('credential_provider')
    except ProfileNotFound:
        # If a user has provided a profile that does not exist,
        # trying to retrieve components/config on the session
        # will raise ProfileNotFound.  Sometimes this is invalid:
        #
        # "ec2 describe-instances --profile unknown"
        #
        # and sometimes this is perfectly valid:
        #
        # "configure set region us-west-2 --profile brand-new-profile"
        #
        # Because we can't know (and don't want to know) whether
        # the customer is trying to do something valid, we just
        # immediately return.  If it's invalid something else
        # up the stack will raise ProfileNotFound, otherwise
        # the configure (and other) commands will work as expected.
        LOG.debug("ProfileNotFound caught when trying to inject "
                  "assume-role cred provider cache.  Not configuring "
                  "JSONFileCache for assume-role.")
        return
    provider = cred_chain.get_provider('assume-role')
    provider.cache = JSONFileCache()

def _run_main(self, parsed_args, parsed_globals):
        # Called when invoked with no args "aws configure"
        new_values = {}
        # This is the config from the config file scoped to a specific
        # profile.
        try:
            config = self._session.get_scoped_config()
        except ProfileNotFound:
            config = {}
        for config_name, prompt_text in self.VALUES_TO_PROMPT:
            current_value = config.get(config_name)
            new_value = self._prompter.get_value(current_value, config_name,
                                                 prompt_text)
            if new_value is not None and new_value != current_value:
                new_values[config_name] = new_value
        config_filename = os.path.expanduser(
            self._session.get_config_variable('config_file'))
        if new_values:
            self._write_out_creds_file_values(new_values,
                                              parsed_globals.profile)
            if parsed_globals.profile is not None:
                new_values['__section__'] = (
                    'profile %s' % parsed_globals.profile)
            self._config_writer.update_config(new_values, config_filename)

def test_aws_instances_profile(mocker):
    test_dir = os.path.dirname(os.path.abspath(__file__))
    cache_dir = os.path.join(test_dir, 'aws_stubs')
    config_dir = os.path.join(test_dir, 'aws_configs')

    aws_obj = AwsInventory(cache_dir=cache_dir, profile='somewhere', region='region', config_path=config_dir)
    instances = aws_obj.instances()

    expected_instances = [Instance(name='test-forms', address='devbox', aliases=('devbox', 'ip-172-31-8-185.us-west-2.compute.internal', 'i-e54cbaeb'), source='aws'),
                          Instance(name='devlab-forms', address='devbox', aliases=('devbox', 'ip-172-31-0-138.us-west-2.compute.internal', 'i-f7d726f9'), source='aws'),
                          Instance(name='test-account-svc', address='devbox', aliases=('devbox', 'ip-172-31-0-139.us-west-2.compute.internal', 'i-f4d726fa'), source='aws'),
                          Instance(name='devlab-pubsrv', address='devbox', aliases=('devbox', 'ip-172-31-0-142.us-west-2.compute.internal', 'i-f5d726fb'), source='aws'),
                          Instance(name='devlab-game-svc', address='devbox', aliases=('devbox', 'ip-172-31-0-140.us-west-2.compute.internal', 'i-f2d726fc'), source='aws'),
                          Instance(name='test-game-svc', address='devbox', aliases=('devbox', 'ip-172-31-0-141.us-west-2.compute.internal', 'i-f3d726fd'), source='aws'),
                          Instance(name='test-pubsrv', address='devbox', aliases=('devbox', 'ip-172-31-2-38.us-west-2.compute.internal', 'i-0f500447384e95942'), source='aws'),
                          Instance(name='test-pubsrv', address='devbox', aliases=('devbox', 'ip-172-31-2-39.us-west-2.compute.internal', 'i-0f500447384e95943'), source='aws')]

    assert set(instances) == set(expected_instances)

    from botocore.exceptions import ProfileNotFound
    with pytest.raises(ProfileNotFound):
        aws_obj = AwsInventory(cache_dir=cache_dir, profile='some-unconfigured-profile', region='region', config_path=config_dir)

def get_scoped_config(self):
        """
        Returns the config values from the config file scoped to the current
        profile.

        The configuration data is loaded **only** from the config file.
        It does not resolve variables based on different locations
        (e.g. first from the session instance, then from environment
        variables, then from the config file).  If you want this lookup
        behavior, use the ``get_config_variable`` method instead.

        Note that this configuration is specific to a single profile (the
        ``profile`` session variable).

        If the ``profile`` session variable is set and the profile does
        not exist in the config file, a ``ProfileNotFound`` exception
        will be raised.

        :raises: ConfigNotFound, ConfigParseError, ProfileNotFound
        :rtype: dict

        """
        profile_name = self.get_config_variable('profile')
        profile_map = self._build_profile_map()
        # If a profile is not explicitly set return the default
        # profile config or an empty config dict if we don't have
        # a default profile.
        if profile_name is None:
            return profile_map.get('default', {})
        elif profile_name not in profile_map:
            # Otherwise if they specified a profile, it has to
            # exist (even if it's the default profile) otherwise
            # we complain.
            raise ProfileNotFound(profile=profile_name)
        else:
            return profile_map[profile_name]

def add_timestamp_parser(session):
    factory = session.get_component('response_parser_factory')
    try:
        timestamp_format = session.get_scoped_config().get(
            'cli_timestamp_format',
            'none')
    except ProfileNotFound:
        # If a --profile is provided that does not exist, loading
        # a value from get_scoped_config will crash the CLI.
        # This function can be called as the first handler for
        # the session-initialized event, which happens before a
        # profile can be created, even if the command would have
        # successfully created a profile. Instead of crashing here
        # on a ProfileNotFound the CLI should just use 'none'.
        timestamp_format = 'none'
    if timestamp_format == 'none':
        # For backwards compatibility reasons, we replace botocore's timestamp
        # parser (which parses to a datetime.datetime object) with the
        # identity function which prints the date exactly the same as it comes
        # across the wire.
        timestamp_parser = identity
    elif timestamp_format == 'iso8601':
        timestamp_parser = iso_format
    else:
        raise ValueError('Unknown cli_timestamp_format value: %s, valid values'
                         ' are "none" or "iso8601"' % timestamp_format)
    factory.set_parser_defaults(timestamp_parser=timestamp_parser)

def get_scoped_config(self):
        """
        Returns the config values from the config file scoped to the current
        profile.

        The configuration data is loaded **only** from the config file.
        It does not resolve variables based on different locations
        (e.g. first from the session instance, then from environment
        variables, then from the config file).  If you want this lookup
        behavior, use the ``get_config_variable`` method instead.

        Note that this configuration is specific to a single profile (the
        ``profile`` session variable).

        If the ``profile`` session variable is set and the profile does
        not exist in the config file, a ``ProfileNotFound`` exception
        will be raised.

        :raises: ConfigNotFound, ConfigParseError, ProfileNotFound
        :rtype: dict

        """
        profile_name = self.get_config_variable('profile')
        profile_map = self._build_profile_map()
        # If a profile is not explicitly set return the default
        # profile config or an empty config dict if we don't have
        # a default profile.
        if profile_name is None:
            return profile_map.get('default', {})
        elif profile_name not in profile_map:
            # Otherwise if they specified a profile, it has to
            # exist (even if it's the default profile) otherwise
            # we complain.
            raise ProfileNotFound(profile=profile_name)
        else:
            return profile_map[profile_name]

def get_scoped_config(self):
        """
        Returns the config values from the config file scoped to the current
        profile.

        The configuration data is loaded **only** from the config file.
        It does not resolve variables based on different locations
        (e.g. first from the session instance, then from environment
        variables, then from the config file).  If you want this lookup
        behavior, use the ``get_config_variable`` method instead.

        Note that this configuration is specific to a single profile (the
        ``profile`` session variable).

        If the ``profile`` session variable is set and the profile does
        not exist in the config file, a ``ProfileNotFound`` exception
        will be raised.

        :raises: ConfigNotFound, ConfigParseError, ProfileNotFound
        :rtype: dict

        """
        profile_name = self.get_config_variable('profile')
        profile_map = self._build_profile_map()
        # If a profile is not explicitly set return the default
        # profile config or an empty config dict if we don't have
        # a default profile.
        if profile_name is None:
            return profile_map.get('default', {})
        elif profile_name not in profile_map:
            # Otherwise if they specified a profile, it has to
            # exist (even if it's the default profile) otherwise
            # we complain.
            raise ProfileNotFound(profile=profile_name)
        else:
            return profile_map[profile_name]

def get_scoped_config(self):
        """
        Returns the config values from the config file scoped to the current
        profile.

        The configuration data is loaded **only** from the config file.
        It does not resolve variables based on different locations
        (e.g. first from the session instance, then from environment
        variables, then from the config file).  If you want this lookup
        behavior, use the ``get_config_variable`` method instead.

        Note that this configuration is specific to a single profile (the
        ``profile`` session variable).

        If the ``profile`` session variable is set and the profile does
        not exist in the config file, a ``ProfileNotFound`` exception
        will be raised.

        :raises: ConfigNotFound, ConfigParseError, ProfileNotFound
        :rtype: dict

        """
        profile_name = self.get_config_variable('profile')
        profile_map = self._build_profile_map()
        # If a profile is not explicitly set return the default
        # profile config or an empty config dict if we don't have
        # a default profile.
        if profile_name is None:
            return profile_map.get('default', {})
        elif profile_name not in profile_map:
            # Otherwise if they specified a profile, it has to
            # exist (even if it's the default profile) otherwise
            # we complain.
            raise ProfileNotFound(profile=profile_name)
        else:
            return profile_map[profile_name]

def get_scoped_config(self):
        """
        Returns the config values from the config file scoped to the current
        profile.

        The configuration data is loaded **only** from the config file.
        It does not resolve variables based on different locations
        (e.g. first from the session instance, then from environment
        variables, then from the config file).  If you want this lookup
        behavior, use the ``get_config_variable`` method instead.

        Note that this configuration is specific to a single profile (the
        ``profile`` session variable).

        If the ``profile`` session variable is set and the profile does
        not exist in the config file, a ``ProfileNotFound`` exception
        will be raised.

        :raises: ConfigNotFound, ConfigParseError, ProfileNotFound
        :rtype: dict

        """
        profile_name = self.get_config_variable('profile')
        profile_map = self._build_profile_map()
        # If a profile is not explicitly set return the default
        # profile config or an empty config dict if we don't have
        # a default profile.
        if profile_name is None:
            return profile_map.get('default', {})
        elif profile_name not in profile_map:
            # Otherwise if they specified a profile, it has to
            # exist (even if it's the default profile) otherwise
            # we complain.
            raise ProfileNotFound(profile=profile_name)
        else:
            return profile_map[profile_name]

def get_scoped_config(self):
        """
        Returns the config values from the config file scoped to the current
        profile.

        The configuration data is loaded **only** from the config file.
        It does not resolve variables based on different locations
        (e.g. first from the session instance, then from environment
        variables, then from the config file).  If you want this lookup
        behavior, use the ``get_config_variable`` method instead.

        Note that this configuration is specific to a single profile (the
        ``profile`` session variable).

        If the ``profile`` session variable is set and the profile does
        not exist in the config file, a ``ProfileNotFound`` exception
        will be raised.

        :raises: ConfigNotFound, ConfigParseError, ProfileNotFound
        :rtype: dict

        """
        profile_name = self.get_config_variable('profile')
        profile_map = self._build_profile_map()
        # If a profile is not explicitly set return the default
        # profile config or an empty config dict if we don't have
        # a default profile.
        if profile_name is None:
            return profile_map.get('default', {})
        elif profile_name not in profile_map:
            # Otherwise if they specified a profile, it has to
            # exist (even if it's the default profile) otherwise
            # we complain.
            raise ProfileNotFound(profile=profile_name)
        else:
            return profile_map[profile_name]

def setup_aws_credentials():
    try:
        # Use existing credentials
        boto3.Session(profile_name=aws_profile_name)
    except ProfileNotFound:
        # Setup new credentials
        print(
            'AWS credentials not found. Please create an IAM user with '
            'programmatic access and AdministratorAccess policy at '
            'https://console.aws.amazon.com/iam/ (On the "Set permissions" '
            'page, choose "Attach existing policies directly" and then select '
            '"AdministratorAccess" policy). After creating the IAM user, '
            'please enter the user\'s Access Key ID and Secret Access '
            'Key below:'
        )
        aws_access_key_id = input('Access Key ID: ')
        aws_secret_access_key = input('Secret Access Key: ')
        if not os.path.exists(os.path.expanduser('~/.aws/')):
            os.makedirs(os.path.expanduser('~/.aws/'))
        aws_credentials_file_path = '~/.aws/credentials'
        aws_credentials_file_string = None
        expanded_aws_file_path = os.path.expanduser(aws_credentials_file_path)
        if os.path.exists(expanded_aws_file_path):
            with open(expanded_aws_file_path, 'r') as aws_credentials_file:
                aws_credentials_file_string = aws_credentials_file.read()
        with open(expanded_aws_file_path, 'a+') as aws_credentials_file:
            # Clean up file
            if aws_credentials_file_string:
                if aws_credentials_file_string.endswith("\n\n"):
                    pass
                elif aws_credentials_file_string.endswith("\n"):
                    aws_credentials_file.write("\n")
                else:
                    aws_credentials_file.write("\n\n")
            # Write login details
            aws_credentials_file.write('[{}]\n'.format(aws_profile_name))
            aws_credentials_file.write(
                'aws_access_key_id={}\n'.format(aws_access_key_id)
            )
            aws_credentials_file.write(
                'aws_secret_access_key={}\n'.format(aws_secret_access_key)
            )
        print('AWS credentials successfully saved in {} file.\n'.format(
            aws_credentials_file_path
        ))
    os.environ['AWS_PROFILE'] = aws_profile_name

def inventory(config):
    inventorySet = InventorySet()

    for source, srcCfg in config.sources():
        if source == 'aws':
            # the cache directory for the original v1 config did not separate 
            # out multiple aws profiles into subdirectories
            if config.version == 1:
                cache_dir = config.inventoryDir(AwsInventory.name)
            else:
                cache_dir = config.inventoryDir(AwsInventory.name, srcCfg['name'])

            if not os.path.exists(cache_dir):
                os.mkdir(cache_dir)

            from botocore.exceptions import ProfileNotFound
            try:
                inv = AwsInventory(cache_dir, **srcCfg)
                inventorySet.add(inv)
            except ProfileNotFound:
                logger.error("Unconfigured AWS profile configured.")
                sys.exit(1)

        elif source == 'csv':
            inv = CsvInventory(path=config.inventoryDir(source, srcCfg['file']), **srcCfg)
            inventorySet.add(inv)

        elif source == 'newrelic':

            proxies = {}

            if config.dig('inventory', 'http_proxy'):
                proxies['http'] = config.dig('inventory', 'http_proxy')
            elif 'HTTP_PROXY' in os.environ:
                proxies['http'] = os.environ['HTTP_PROXY']
            elif 'http_proxy' in os.environ:
                proxies['http'] = os.environ['http_proxy']

            if config.dig('inventory', 'https_proxy'):
                proxies['https'] = config.dig('inventory', 'https_proxy')
            elif 'HTTPS_PROXY' in os.environ:
                proxies['https'] = os.environ['HTTPS_PROXY']
            elif 'https_proxy' in os.environ:
                proxies['https'] = os.environ['https_proxy']

            inv = NewRelicInventory(data_path=config.inventoryDir(NewRelicInventory.name),
                                    proxies=proxies,
                                    **srcCfg)

            inventorySet.add(inv)

    return inventorySet