feat: lambda for put-timetable

src/lambda/put-timetable/index.py

@@ -0,0 +1,32 @@
+import json
+from datetime import datetime
+from utils import JsonPayloadBuilder
+from utils import resp_handler
+from utils import table
+
+@resp_handler
+def put_timetable(uid, content):
+    dt_now = datetime.now().strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'
+    table.update_item(
+        Key = {
+            "uid": uid
+        },
+        UpdateExpression = 'SET courses = :info, update_at = :ts',
+        ExpressionAtrributeValues = {
+            ":info" : [content['courses']],
+            ":ts" : dt_now
+        }
+    )
+
+    body = JsonPayloadBuilder().add_status(True).add_data(None).add_message('').compile()
+    return body
+
+
+def handler(event, context):
+    req = json.loads(event['body'])
+    params = {
+        "uid": event['requestContext']['authorizer']['claims']['sub'],
+        "content": req["data"]
+    }
+
+    return put_timetable(**params)   

src/lambda/put-timetable/utils.py

@@ -0,0 +1,66 @@
+import boto3
+import json
+import logging
+import os
+from decimal import Decimal
+
+db = boto3.resource("dynamodb", region_name="ap-northeast-1")
+table = db.Table(os.getenv('TABLE_NAME'))
+
+
+class DecimalEncoder(json.JSONEncoder):
+    def default(self, obj):
+        if isinstance(obj, Decimal):
+            return float(obj)
+        return json.JSONEncoder.default(self, obj)
+
+
+class JsonPayloadBuilder:
+    payload = {}
+
+    def add_status(self, success):
+        self.payload['success'] = success
+        return self
+
+    def add_data(self, data):
+        self.payload['data'] = data
+        return self
+
+    def add_message(self, msg):
+        self.payload['message'] = msg
+        return self
+
+    def compile(self):
+        return json.dumps(self.payload, cls=DecimalEncoder, ensure_ascii=False).encode('utf8')
+
+
+def api_response(code, body):
+    return {
+        "isBase64Encoded": False,
+        "statusCode": code,
+        'headers': {
+            "Access-Control-Allow-Origin": '*',
+            "Content-Type": "application/json",
+            "Referrer-Policy": "origin"
+        },
+        "multiValueHeaders": {"Access-Control-Allow-Methods": ["POST", "OPTIONS", "GET", "PATCH", "DELETE", "PUT"]},
+        "body": body
+    }
+
+
+def resp_handler(func):
+    def handle(*args, **kwargs):
+        try:
+            resp = func(*args, **kwargs)
+            return api_response(200, resp)
+        except LookupError:
+            resp = JsonPayloadBuilder().add_status(False).add_data(None) \
+                .add_message("Not found").compile()
+            return api_response(404, resp)
+        except Exception as e:
+            logging.error(str(e))
+            resp = JsonPayloadBuilder().add_status(False).add_data(None) \
+                .add_message("Internal error, please contact [email protected].").compile()
+            return api_response(500, resp)
+
+    return handle