def _get_resources_from_account()

in src/inventory/readers.py [0:0]

• 22 lines of code
• 4 McCabe index (conditional complexity)

    def _get_resources_from_account(self, account_id: str) -> Iterator[List[str]]:
        try:
            _logger.info(f"assuming role on account {account_id}")

            sts_response = self._sts_client.assume_role(RoleArn=f"arn:{self._get_aws_partition()}:iam::{account_id}:role/{os.environ['CROSS_ACCOUNT_ROLE_NAME']}",
                                                        RoleSessionName=f"{account_id}-Assumed-Role",
                                                        DurationSeconds=900)
            config_client = self._get_config_client(sts_response)

            next_token: str = ''
            while True:
                resources_result = config_client.select_resource_config(Expression="SELECT arn, resourceType, configuration, tags "
                                                                                   "WHERE resourceType IN ('AWS::EC2::Instance', 'AWS::ElasticLoadBalancingV2::LoadBalancer', "
                                                                                       "'AWS::ElasticLoadBalancing::LoadBalancer', 'AWS::DynamoDB::Table', 'AWS::RDS::DBInstance')",
                                                                        NextToken=next_token)
                
                next_token = resources_result.get('NextToken', '')
                results: List[str] = resources_result.get('Results', [])

                _logger.debug(f"page returned {len(results)} and next token of '{next_token}'")

                yield results

                if not next_token:
                    break
        except ClientError as ex:
            _logger.error("Received error: %s while retrieving resources from account %s, moving onto next account.", ex, account_id, exc_info=True)

            yield []