def retrieve_stack()

in awscli/customizations/opsworks.py [0:0]

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

    def retrieve_stack(self, args):
        """
        Retrieves the stack from the API, thereby ensures that it exists.

        Provides `self._stack`, `self._prov_params`, `self._use_address`, and
        `self._ec2_instance`.
        """

        LOG.debug("Retrieving stack and provisioning parameters")
        self._stack = self.opsworks.describe_stacks(
            StackIds=[args.stack_id]
        )['Stacks'][0]
        self._prov_params = \
            self.opsworks.describe_stack_provisioning_parameters(
                StackId=self._stack['StackId']
            )

        if args.infrastructure_class == 'ec2' and not args.local:
            LOG.debug("Retrieving EC2 instance information")
            ec2 = self._session.create_client(
                'ec2', region_name=self._stack['Region'])

            # `desc_args` are arguments for the describe_instances call,
            # whereas `conditions` is a list of lambdas for further filtering
            # on the results of the call.
            desc_args = {'Filters': []}
            conditions = []

            # make sure that the platforms (EC2/VPC) and VPC IDs of the stack
            # and the instance match
            if 'VpcId' in self._stack:
                desc_args['Filters'].append(
                    {'Name': 'vpc-id', 'Values': [self._stack['VpcId']]}
                )
            else:
                # Cannot search for non-VPC instances directly, thus filter
                # afterwards
                conditions.append(lambda instance: 'VpcId' not in instance)

            # target may be an instance ID, an IP address, or a name
            if INSTANCE_ID_RE.match(args.target):
                desc_args['InstanceIds'] = [args.target]
            elif IP_ADDRESS_RE.match(args.target):
                # Cannot search for either private or public IP at the same
                # time, thus filter afterwards
                conditions.append(
                    lambda instance:
                        instance.get('PrivateIpAddress') == args.target or
                        instance.get('PublicIpAddress') == args.target)
                # also use the given address to connect
                self._use_address = args.target
            else:
                # names are tags
                desc_args['Filters'].append(
                    {'Name': 'tag:Name', 'Values': [args.target]}
                )

            # find all matching instances
            instances = [
                i
                for r in ec2.describe_instances(**desc_args)['Reservations']
                for i in r['Instances']
                if all(c(i) for c in conditions)
            ]

            if not instances:
                raise ValueError(
                    "Did not find any instance matching %s." % args.target)
            elif len(instances) > 1:
                raise ValueError(
                    "Found multiple instances matching %s: %s." % (
                        args.target,
                        ", ".join(i['InstanceId'] for i in instances)))

            self._ec2_instance = instances[0]