def assume_roles_chain()

in modular_sdk/services/sts_service.py [0:0]

• 33 lines of code
• 5 McCabe index (conditional complexity)

    def assume_roles_chain(self, payloads: List[AssumeRolePayload]
                           ) -> AssumeRoleResult:
        """
        It assumes a chain of roles one after another.
        :param payloads:
        :return: AssumeRoleResult
        Returns the credentials and expiration of the last assumed role
        """
        assert payloads, 'At least one payload must be given'
        _sts = self.client
        for payload in payloads:
            assert len(payload) == 3, 'Invalid usage of the method'
            arn = payload[0]
            session_name = payload[1] or f'modular_sdk-sdk-session-{time()}'
            duration = payload[2] or 900
            try:
                _LOG.info(f'Assuming {arn} from chain')
                creds = _sts.assume_role(
                    RoleArn=arn,
                    RoleSessionName=session_name,
                    DurationSeconds=duration
                )['Credentials']
            except ClientError as e:
                error_message = f'Error while assuming {arn} from chain'
                _LOG.error(f'{error_message}: {e}')
                raise ConnectionAbortedError(error_message) from e
            _sts = AWSCredentialsProvider(
                service_name='sts',
                aws_region=self._region_name,
                aws_access_key_id=creds['AccessKeyId'],
                aws_secret_access_key=creds['SecretAccessKey'],
                aws_session_token=creds['SessionToken'],
            ).client
        # creds variable will exist, ignore warning
        return {
            'aws_access_key_id': creds.get('AccessKeyId'),
            'aws_secret_access_key': creds.get('SecretAccessKey'),
            'aws_session_token': creds.get('SessionToken'),
            'expiration': creds.get('Expiration')  # datetime UTC
        }