create

handler_c_contact.py

@@ -0,0 +1,26 @@
+import json
+import logging
+import urllib
+from datetime import datetime
+
+from src.create_signup import CreateSignup
+from src.response import Response
+
+
+def handle(event, context):
+    try:
+        event_body = json.loads(urllib.parse.unquote_plus(event["body"]))
+
+        signs = CreateSignup()
+        create = signs.create(
+            event["requestContext"]["authorizer"]["claims"]["email"],
+            datetime.now(),
+            event_body[0],
+        )
+
+        return Response.handle(create, 200)
+
+    except Exception as e:
+        msg = f"Unable{str(e)}"
+        logging.exception(msg)
+        return Response.handle({"error": msg}, 500)

src/create_signup.py

@@ -0,0 +1,37 @@
+import uuid
+from datetime import datetime
+from multiprocessing import Process
+
+import boto3
+
+
+class CreateSignup:
+    def __init__(self):
+        self.dynamodb = boto3.resource("dynamodb", region_name="eu-west-1")
+        self.table = self.dynamodb.Table("signups")
+
+    def create(
+        self, email: str, request_datetime: datetime, full_name: str
+    ) -> dict:
+
+        signup = self.createSignupObject(full_name, request_datetime)
+
+        p = Process(target=self.c, args=(email, full_name, request_datetime))
+        p.start()
+
+    def c(self, email, full_name, request_datetime):
+        self.table.put_item(
+            Item={
+                'email': email,
+                "uuid": str(uuid.uuid4()),
+                "name": full_name,
+                "time": request_datetime.strftime("%H:%M:%S"),
+            }
+        )
+
+    def createSignupObject(self, full_name, request_datetime):
+        return {
+            "uuid": str(uuid.uuid4()),
+            "name": full_name,
+            "time": request_datetime.strftime("%H:%M:%S"),
+        }

src/response.py

@@ -0,0 +1,14 @@
+import json
+
+
+class Response:
+    @staticmethod
+    def handle(message, status_code=200):
+        return {
+            "statusCode": str(status_code),
+            "body": json.dumps(message),
+            "headers": {
+                "Content-Type": "application/json",
+                "Access-Control-Allow-Origin": "*",
+            },
+        }

template.yaml

@@ -11,6 +11,57 @@ Globals:
     Timeout: 3
 
 Resources:
+
+  SignupApi:
+    Type: AWS::Serverless::Api
+    Properties:
+      Name: SignupApo
+      StageName: test
+      Cors:
+        AllowMethods: "'*'"
+        AllowHeaders: "'*'"
+        AllowOrigin: "'*'"
+
+  SignupTable:
+    Type: AWS::DynamoDB::Table
+    Properties: 
+      TableName: "signups"
+      AttributeDefinitions: 
+        - 
+          AttributeName: "email"
+          AttributeType: "S"
+        - 
+          AttributeName: "day"
+          AttributeType: "S"
+      KeySchema: 
+        - 
+          AttributeName: "email"
+          KeyType: "HASH"
+        - 
+          AttributeName: "day"
+          KeyType: "RANGE"
+      ProvisionedThroughput: 
+        ReadCapacityUnits: 5
+        WriteCapacityUnits: 5
+
+
+
+        ContactWrite:
+          Type: AWS::Serverless::Function
+          Properties:
+            CodeUri: .
+            Handler: handler_c_contact.h
+            Runtime: python3.7
+            Policies:
+              - AmazonDynamoDBFullAccess
+            Events:
+              Write:
+                Type: Api
+                Properties:
+                  Path: /write
+                  RestApiId: !Ref SignupApi
+                  Method: POST
+
   HelloWorldFunction:
     Type: AWS::Serverless::Function # More info about Function Resource: https://github.com/awslabs/serverless-application-model/blob/master/versions/2016-10-31.md#awsserverlessfunction
     Properties:

tests/unit/test_create_signup.py

@@ -0,0 +1,33 @@
+import unittest
+from datetime import datetime
+
+import boto3
+from moto import mock_dynamodb2
+
+from src.create_signup import CreateSignup
+
+
+class TestCreateSignup(unittest.TestCase):
+    @mock_dynamodb2
+    def test_creates_signup(self):
+
+        dynamodb = boto3.resource("dynamodb", region_name="eu-west-1")
+
+        table = dynamodb.create_table(
+            TableName="contacts",
+            KeySchema=[
+                {"AttributeName": "email", "KeyType": "HASH"},
+                {"AttributeName": "day", "KeyType": "RANGE"},
+            ],
+            AttributeDefinitions=[
+                {"AttributeName": "email", "AttributeType": "S"},
+                {"AttributeName": "day", "AttributeType": "S"},
+            ],
+            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+        )
+
+        client = CreateSignup()
+        response = client.create('[email protected]', '11/10/2020', 'Stu Mason')
+
+
+        print(result)

tests/unit/test_response.py

@@ -0,0 +1,8 @@
+import unittest
+
+from src.response import Response
+
+
+class TestResponse(unittest.TestCase):
+    def test_response(self):
+        Response.handle("Foo", 200)