def classify_email_lambda()

in stacks/email_automation_workflow_stack.py [0:0]

• 39 lines of code
• 1 McCabe index (conditional complexity)

    def classify_email_lambda(self, human_workflow_topic):
        
        email_classification_endpoint_arn = core.CfnParameter(self, "emailClassificationEndpointArn",
                #type="String"
                ).value_as_string
                
        email_entity_recognition_endpoint_arn = core.CfnParameter(self, "emailEntityRecognitionEndpointArn",
                #type="String"
                ).value_as_string
                
        support_email = core.CfnParameter(self, "supportEmail",
                #type="String"
                ).value_as_string
        
        email_classify_lambda = lambda_.Function(
            self, "id_classify_emails_lambda_fn", 
            function_name="classify-emails-lambda-fn",
            code = lambda_.Code.from_asset(path.join("./lambda", "classify-emails-lambda")),
            handler = "lambda_function.lambda_handler",
            runtime = lambda_.Runtime.PYTHON_3_8,
            timeout = core.Duration.minutes(1),
            environment={
                "EMAIL_CLASSIFICATION_ENDPOINT_ARN" : email_classification_endpoint_arn,
                "EMAIL_ENTITY_RECOGNITION_ENDPOINT_ARN" : email_entity_recognition_endpoint_arn,
                "HUMAN_WORKFLOW_SNS_TOPIC_ARN" : human_workflow_topic.topic_arn,
                "SOURCE_EMAIL" : support_email
            }
        )
        
        email_classify_lambda.add_to_role_policy(
            iam.PolicyStatement(
                        actions = [
                            "comprehend:*",
                        ],
                        resources= [ email_classification_endpoint_arn, email_entity_recognition_endpoint_arn ]
                    )
            
        )

        email_classify_lambda.add_to_role_policy(
            iam.PolicyStatement(
                        actions = [
                            "ses:SendTemplatedEmail",
                        ],
                        resources= [ '*' ]
                    )
            
        )
        
        human_workflow_topic.grant_publish(email_classify_lambda)
        
        return email_classify_lambda