Add serverless config

.dockerignore

@@ -0,0 +1,8 @@
+__pycache__/
+.git/
+.gitignore
+
+*.env
+
+.DS_Store
+.vscode/

Dockerfile

@@ -1,10 +1,25 @@
-FROM python:3.9-slim
+FROM python:3.9-alpine
 
 WORKDIR /app
 
+RUN apk add --no-cache \
+    nodejs \
+    npm \
+    bash
+
 COPY requirements.txt /app/requirements.txt
 RUN pip install --no-cache-dir -r requirements.txt
 
+# Copy Serverless config before installing plugins
+COPY serverless.yml /app/serverless.yml
+
+# Install Serverless Framework globally
+RUN npm install -g serverless
+
+# Install Serverless plugins
+RUN serverless plugin install -n serverless-wsgi && \
+    serverless plugin install -n serverless-python-requirements
+
 COPY . /app
 
 EXPOSE 5000

api/app.py

@@ -19,7 +19,7 @@ def health_check():
 def create_user():
     data = request.json
     user = {"user_id": data["user_id"], "name": data["name"], "email": data["email"]}
-
+    print("USER DATA app", user)
     try:
         UserModel.create(user)
         return jsonify({"message": "User created", "user_id": data["user_id"]}), 201

api/dynamodb.py

@@ -7,24 +7,30 @@
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)
 
-TABLE_NAME = "UserTable"
-
-
-dynamodb_client = boto3.client(
-    "dynamodb",
-    endpoint_url=os.environ.get("DYNAMODB_HOST", None),
-    region_name=os.environ["AWS_REGION"],
-    aws_access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
-    aws_secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
-)
-
-dynamodb_resource = boto3.resource(
-    "dynamodb",
-    endpoint_url=os.environ.get("DYNAMODB_HOST", None),
-    region_name=os.environ["AWS_REGION"],
-    aws_access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
-    aws_secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
-)
+TABLE_NAME = os.environ.get("TABLE_NAME", "user_table")
+IS_OFFLINE = os.environ.get("IS_OFFLINE")
+
+if IS_OFFLINE:
+    # For local development AWS credentials need to be passed, but values can be faked
+    dynamodb_client = boto3.client(
+        "dynamodb",
+        endpoint_url=os.environ.get("DYNAMODB_HOST", None),
+        region_name=os.environ.get("AWS_REGION", "us-east-1"),
+        aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID", "fake-key"),
+        aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY", "fake-secret"),
+    )
+
+    dynamodb_resource = boto3.resource(
+        "dynamodb",
+        endpoint_url=os.environ.get("DYNAMODB_HOST", None),
+        region_name=os.environ.get("AWS_REGION", "us-east-1"),
+        aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID", "fake-key"),
+        aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY", "fake-secret"),
+    )
+else:
+    # For AWS Lambda execution
+    dynamodb_client = boto3.client("dynamodb")
+    dynamodb_resource = boto3.resource("dynamodb")
 
 
 def create_user_table():
@@ -44,8 +50,8 @@ def create_user_table():
         dynamodb_client.create_table(
             TableName=TABLE_NAME,
             KeySchema=[
-                {"AttributeName": "user_id", "KeyType": "HASH"}  # Partition key
-            ],
+                {"AttributeName": "user_id", "KeyType": "HASH"}
+            ],  # Partition key
             AttributeDefinitions=[
                 {"AttributeName": "user_id", "AttributeType": "S"}  # String
             ],
@@ -75,5 +81,7 @@ def get_user_table():
 
 
 def create_tables():
-    create_user_table()
-    logger.info("All tables created successfully")
+    if IS_OFFLINE:
+        create_user_table()
+        logger.info("All tables created successfully")
+    return

api/models.py

@@ -6,46 +6,38 @@
 class UserModel:
     @staticmethod
     def get(user_id):
-        """Get a user by user_id"""
         table = get_user_table()
         try:
             response = table.get_item(Key={"user_id": user_id})
             if "Item" not in response:
                 raise Exception("DoesNotExist")
             return response["Item"]
-        except ClientError as e:
-            print(f"Error getting user: {e}")
+        except ClientError:
             raise
 
     @staticmethod
     def create(user_data):
-        """Create a new user"""
         table = get_user_table()
         try:
             table.put_item(Item=user_data)
             return user_data
-        except ClientError as e:
-            print(f"Error creating user: {e}")
+        except ClientError:
             raise
 
     @staticmethod
     def scan():
-        """Scan all users"""
         table = get_user_table()
         try:
             response = table.scan()
             return response.get("Items", [])
-        except ClientError as e:
-            print(f"Error scanning users: {e}")
+        except ClientError:
             raise
 
     @staticmethod
     def delete(user_id):
-        """Delete a user"""
         table = get_user_table()
         try:
             response = table.delete_item(Key={"user_id": user_id})
             return response
-        except ClientError as e:
-            print(f"Error deleting user: {e}")
+        except ClientError:
             raise

docker-compose.yml

@@ -14,7 +14,8 @@ services:
       - AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}
       - SERVERLESS_ACCESS_KEY=${SERVERLESS_ACCESS_KEY}
       - PORT=${PORT}
-      - DYNAMODB_HOST=${DYNAMODB_HOST}
+      - DYNAMODB_HOST=http://dynamodb:8000
+      - IS_OFFLINE=1
     depends_on:
       - dynamodb
 

env.example

@@ -0,0 +1,9 @@
+# Valid AWS credentials are required for serverless deploy
+AWS_REGION=your-aws-region
+AWS_ACCESS_KEY_ID=your-access-key
+AWS_SECRET_ACCESS_KEY=your-secret-key
+SERVERLESS_ACCESS_KEY=your-serverless-key
+
+# Application related variables
+PORT=5000
+FLASK_DEBUG=1

serverless.yml

@@ -0,0 +1,64 @@
+service: dynamodb-flask-lambda
+
+provider:
+  name: aws
+  runtime: python3.9
+  region: ${env:AWS_REGION}
+  memorySize: 128
+  iamRoleStatements:
+  - Effect: Allow
+    Action:
+    - dynamodb:Query
+    - dynamodb:Scan
+    - dynamodb:GetItem
+    - dynamodb:PutItem
+    - dynamodb:UpdateItem
+    - dynamodb:DeleteItem
+    Resource:
+    - { "Fn::GetAtt": [ "UserTable", "Arn" ] }
+plugins:
+- serverless-python-requirements
+- serverless-wsgi
+package:
+  exclude:
+  - node_modules/**
+  - tests/**
+  - .gitignore
+  - .github/
+  - requirements-lint.txt
+  - docker-compose.yml
+  - Dockerfile
+  - .env
+  include:
+  - api/templates/**
+custom:
+  wsgi:
+    app: api.app.app
+    pythonRequirements:
+      dockerizePip: true
+functions:
+  app:
+    handler: wsgi_handler.handler
+    environment:
+      FLASK_DEBUG: 0
+      TABLE_NAME:
+        Ref: UserTable
+    events:
+    - http: ANY /
+    - http: 'ANY {proxy+}'
+resources:
+  Resources:
+    UserTable:
+      Type: 'AWS::DynamoDB::Table'
+      Properties:
+        AttributeDefinitions:
+
+        - AttributeName: user_id
+          AttributeType: S
+        KeySchema:
+
+        - AttributeName: user_id
+          KeyType: HASH
+        ProvisionedThroughput:
+          ReadCapacityUnits: 1
+          WriteCapacityUnits: 1

tests/test_dynamodb.py

@@ -13,7 +13,7 @@ def test_create_tables_already_exists(self, setup_dynamodb_mock):
 
         create_tables()
 
-        mock_client.describe_table.assert_called_once_with(TableName="UserTable")
+        mock_client.describe_table.assert_called_once_with(TableName="user_table")
         mock_client.create_table.assert_not_called()
 
     def test_create_tables_new(self, setup_dynamodb_mock):
@@ -29,20 +29,20 @@ def test_create_tables_new(self, setup_dynamodb_mock):
 
         create_tables()
 
-        mock_client.describe_table.assert_called_once_with(TableName="UserTable")
+        mock_client.describe_table.assert_called_once_with(TableName="user_table")
         mock_client.create_table.assert_called_once()
-        mock_waiter.wait.assert_called_once_with(TableName="UserTable")
+        mock_waiter.wait.assert_called_once_with(TableName="user_table")
 
     def test_list_tables(self, setup_dynamodb_mock):
         mock_client, _, _ = setup_dynamodb_mock
-        mock_client.list_tables.return_value = {"TableNames": ["UserTable"]}
+        mock_client.list_tables.return_value = {"TableNames": ["user_table"]}
 
         from api.dynamodb import list_tables
 
         result = list_tables()
 
         mock_client.list_tables.assert_called_once()
-        assert result == ["UserTable"]
+        assert result == ["user_table"]
 
     def test_get_user_table(self, setup_dynamodb_mock):
         _, mock_resource, mock_table = setup_dynamodb_mock
@@ -51,5 +51,5 @@ def test_get_user_table(self, setup_dynamodb_mock):
 
         result = get_user_table()
 
-        mock_resource.Table.assert_called_once_with("UserTable")
+        mock_resource.Table.assert_called_once_with("user_table")
         assert result == mock_table