def convert()

in syndicate/core/transform/terraform/converter/tf_lambda_converter.py [0:0]

• 62 lines of code
• 8 McCabe index (conditional complexity)

    def convert(self, name, resource):
        validate_params(name, resource, REQUIRED_PARAMS)

        function_name = resource.get('func_name')
        runtime = resource.get('runtime')
        memory = resource.get('memory')
        timeout = resource.get('timeout')
        env_variables = resource.get('env_variables')
        publish_version = resource.get('publish_version', False)
        subnet_ids = resource.get('subnet_ids', [])
        security_group_ids = resource.get('security_group_ids', [])

        s3_path = resource.get('s3_path')
        s3_bucket = self.config.deploy_target_bucket

        iam_role_name = resource.get('iam_role_name')
        iam_conn = self.resources_provider.iam().iam_conn
        role_arn = iam_conn.check_if_role_exists(iam_role_name)
        if not role_arn:
            iam_role = self.template.get_resource_by_name(iam_role_name)
            if not iam_role:
                raise AssertionError(f'Role {iam_role_name} does not exist; '
                                     f'Lambda {name} failed to be configured.')
            role_arn = build_role_arn_ref(iam_role_name)

        lambda_layers_arns = []
        layer_meta = resource.get('layers')
        if layer_meta:
            for layer_name in layer_meta:
                layer = self.template.get_resource_by_name(
                    lambda_layer_name(layer_name=layer_name))
                if not layer:
                    raise AssertionError(
                        f"Lambda layer '{layer_name}' is not present "
                        "in build meta.")
                layer_ref = build_ref_to_lambda_layer_arn(
                    layer_name=layer_name)
                lambda_layers_arns.append(layer_ref)

        retention = resource.get('logs_expiration')
        if retention:
            log_group = cloud_watch_log_group(lambda_name=function_name,
                                              retention=retention)
            self.template.add_cloud_watch_log_group(meta=log_group)

        reserved_concur = resource.get(LAMBDA_MAX_CONCURRENCY)
        lambda_service = self.resources_provider.lambda_resource()
        if not lambda_service.check_concurrency_availability(reserved_concur):
            reserved_concur = None

        aws_lambda = template_for_lambda(lambda_name=name,
                                         function_name=name,
                                         runtime=runtime,
                                         role_arn=role_arn,
                                         handler=function_name,
                                         memory=memory,
                                         timeout=timeout,
                                         env_variables=env_variables,
                                         publish=publish_version,
                                         subnet_ids=subnet_ids,
                                         security_group_ids=security_group_ids,
                                         reserved_concurrent_executions=reserved_concur,
                                         s3_bucket=s3_bucket,
                                         s3_key=s3_path)
        self.template.add_aws_lambda(meta=aws_lambda)
        self.configure_concurrency(resource=resource,
                                   function_name=name)
        self.process_event_sources(lambda_name=name, resource=resource,
                                   role=iam_role_name)