def _create_lambda_from_meta()

in syndicate/core/resources/lambda_resource.py [0:0]

• 108 lines of code
• 19 McCabe index (conditional complexity)

    def _create_lambda_from_meta(self, name, meta):
        from syndicate.core import CONFIG
        _LOG.debug('Creating lambda %s', name)
        req_params = ['iam_role_name', 'runtime', 'memory', 'timeout',
                      'func_name']
        # Lambda configuration
        validate_params(name, meta, req_params)

        key = meta[S3_PATH_NAME]
        key_compound = PurePath(CONFIG.deploy_target_bucket_key_compound,
                                key).as_posix()
        if not self.s3_conn.is_file_exists(self.deploy_target_bucket,
                                           key_compound):
            raise AssertionError(f'Error while creating lambda: {name};'
                                 f'Deployment package {key_compound} does not exist '
                                 f'in {self.deploy_target_bucket} bucket')

        lambda_def = self.lambda_conn.get_function(name)
        if lambda_def:
            _LOG.warn('%s lambda exists.', name)
            return self.describe_lambda(name, meta, lambda_def)

        role_name = meta['iam_role_name']
        role_arn = self.iam_conn.check_if_role_exists(role_name)
        if not role_arn:
            raise AssertionError(f'Role {role_name} does not exist; '
                                 f'Lambda {name} failed to be configured.')

        dl_target_arn = self.get_dl_target_arn(meta=meta,
                                               region=self.region,
                                               account_id=self.account_id)

        publish_version = meta.get('publish_version', False)
        lambda_layers_arns = []
        layer_meta = meta.get('layers')
        if layer_meta:
            if 'dotnet' in meta['runtime'].lower():
                env_vars = meta.get('env_variables', {})
                env_vars.update(_DOTNET_LAMBDA_SHARED_STORE_ENV)
                meta['env_variables'] = env_vars
            for layer_name in layer_meta:
                layer_arn = self.lambda_conn.get_lambda_layer_arn(layer_name)
                if not layer_arn:
                    raise AssertionError(
                        'Could not link lambda layer {} to lambda {} '
                        'due to layer absence!'.format(layer_name, name))
                lambda_layers_arns.append(layer_arn)

        ephemeral_storage = meta.get('ephemeral_storage', 512)

        if meta.get('env_variables'):
            self._resolve_env_variables(meta.get('env_variables'))

        self.lambda_conn.create_lambda(
            lambda_name=name,
            func_name=meta['func_name'],
            role=role_arn,
            runtime=meta['runtime'].lower(),
            memory=meta['memory'],
            timeout=meta['timeout'],
            s3_bucket=self.deploy_target_bucket,
            s3_key=key_compound,
            env_vars=meta.get('env_variables'),
            vpc_sub_nets=meta.get('subnet_ids'),
            vpc_security_group=meta.get('security_group_ids'),
            dl_target_arn=dl_target_arn,
            tracing_mode=meta.get('tracing_mode'),
            publish_version=publish_version,
            layers=lambda_layers_arns,
            ephemeral_storage=ephemeral_storage,
            snap_start=self._resolve_snap_start(meta=meta),
            architectures=meta.get('architectures'),
            tags=meta.get('tags')
        )
        _LOG.debug('Lambda created %s', name)
        # AWS sometimes returns None after function creation, needs for
        # stability
        waiter = self.lambda_conn.get_waiter('function_exists')
        waiter.wait(FunctionName=name)

        self._resolve_log_group(lambda_name=name, meta=meta)

        lambda_def = self.__describe_lambda_by_version(
            name) if publish_version else self.lambda_conn.get_function(name)
        version = lambda_def['Configuration']['Version']
        self._setup_function_concurrency(name=name, meta=meta)

        # enabling aliases,
        # aliases can be enabled only and for $LATEST
        alias = meta.get('alias')
        if alias:
            _LOG.debug('Creating alias')
            _LOG.debug(self.lambda_conn.create_alias(function_name=name,
                                                     name=alias,
                                                     version=version))
        url_config = meta.get('url_config')
        if url_config:
            _LOG.info('Url config is found. Setting the function url')
            url = self.lambda_conn.set_url_config(
                function_name=name, auth_type=url_config.get('auth_type'),
                qualifier=alias, cors=url_config.get('cors'),
                principal=url_config.get('principal'),
                source_arn=url_config.get('source_arn')
            )
            USER_LOG.info(f'{name}:{alias if alias else ""} URL: {url}')

        arn = self.build_lambda_arn_with_alias(lambda_def, alias) \
            if publish_version or alias else \
            lambda_def['Configuration']['FunctionArn']
        _LOG.debug(f'Resolved lambda arn: {arn}')

        event_sources_meta = meta.get('event_sources', [])
        self.create_lambda_triggers(name, arn, role_name, event_sources_meta)

        if meta.get('max_retries') is not None:
            _LOG.debug('Setting lambda event invoke config')
            function_name = (name + ":" + alias) if alias else name
            invoke_config = self.lambda_conn.put_function_event_invoke_config(
                function_name=function_name,
                max_retries=meta.get('max_retries')
            )
            _LOG.debug(f'Created lambda invoke config: {invoke_config}')

        # concurrency configuration
        self._manage_provisioned_concurrency_configuration(
            function_name=name,
            meta=meta,
            lambda_def=lambda_def)
        return self.describe_lambda(name, meta, lambda_def)