API with dynamodb back end

Author: Jade Wibbels

.gitignore

@@ -25,6 +25,7 @@ share/python-wheels/
 .installed.cfg
 *.egg
 MANIFEST
+package/
 
 # PyInstaller
 #  Usually these files are written by a python script from a template

app/__init__.py

@@ -0,0 +1,27 @@
+# Description: This script is used to create the tables in the dynamodb
+aws dynamodb create-table \
+    --table-name users \
+    --attribute-definitions AttributeName=Id,AttributeType=S \
+    --key-schema AttributeName=Id,KeyType=HASH \
+    --provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5
+
+aws dynamodb create-table \
+    --table-name posts \
+    --attribute-definitions AttributeName=Id,AttributeType=S \
+    --key-schema AttributeName=Id,KeyType=HASH \
+    --provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5
+
+# {
+#     "ApiEndpoint": "https://x8ttp4bxtb.execute-api.us-east-1.amazonaws.com",
+#     "ApiId": "x8ttp4bxtb",
+#     "ApiKeySelectionExpression": "$request.header.x-api-key",
+#     "CreatedDate": "2024-12-04T01:24:13+00:00",
+#     "DisableExecuteApiEndpoint": false,
+#     "Name": "posts-api",
+#     "ProtocolType": "HTTP",
+#     "RouteSelectionExpression": "$request.method $request.path"
+# }
+aws apigatewayv2 create-api \
+    --name posts-api \
+    --protocol-type HTTP \
+    --target arn:aws:lambda:us-east-1:123456789012:function:posts-api

app/awscli.sh

@@ -0,0 +1,28 @@
+from fastapi import FastAPI
+from mangum import Mangum
+from app.routes.users import router as users_router
+from app.routes.posts import router as posts_router
+
+app = FastAPI()
+
+app.include_router(users_router, prefix="/users", tags=["users"])
+app.include_router(posts_router, prefix="/posts", tags=["posts"])
+
+
+@app.get("/")
+def read_root():
+    """
+    Root endpoint
+    """
+    return {"message": "Welcome to the API"}
+
+
+@app.get("/health")
+def read_health():
+    """
+    Health check endpoint
+    """
+    return {"message": "API is healthy"}
+
+
+handler = Mangum(app, lifespan="off", api_gateway_base_path="/api/v1")

app/main.py

@@ -0,0 +1,129 @@
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
+import boto3
+from botocore.exceptions import ClientError
+
+router = APIRouter()
+
+# Initialize the DynamoDB client
+dynamodb = boto3.resource("dynamodb")
+table = dynamodb.Table("posts")
+
+
+class Post(BaseModel):
+    Id: str
+    title: str
+    content: str
+    author_id: str
+
+
+@router.get("/")
+def read_posts():
+    """
+    Gets all posts from the DynamoDB table "posts"
+
+    Example call:
+    ``curl -X GET "http://127.0.0.1:8000/posts/" -H "accept: application/json"``
+
+    Response:
+    ``[{"Id": "1", "title": "First Post", ... "author_id": "1"}]``
+    """
+    try:
+        response = table.scan()
+        return response["Items"]
+    except ClientError as e:
+        raise HTTPException(status_code=500, detail=e.response["Error"]["Message"])
+
+
+@router.get("/{Id}")
+def read_post(Id: str):
+    """
+    Gets a post from the DynamoDB table "posts" by Id
+
+    Example call:
+    ``curl -X GET "http://127.0.0.1:8000/posts/1" -H "accept: application/json"``
+
+    Response:
+    ``{"content":"This is the first post","author_id":"1","Id":"1","title":"First Post"}``
+    """
+    try:
+        response = table.get_item(Key={"Id": Id})
+        if "Item" not in response:
+            raise HTTPException(status_code=404, detail="Post not found")
+        return response["Item"]
+    except ClientError as e:
+        raise HTTPException(status_code=500, detail=e.response["Error"]["Message"])
+
+
+@router.post("/")
+def create_post(post: Post):
+    """
+    Creates a post in the DynamoDB table "posts"
+
+    Example call:
+    ``curl -X POST "http://127.0.0.1:8000/posts/" -H "accept: application/json" -H "Content-Type: application/json" -d '{  # noqa E501
+      "Id": "1",
+      "title": "First Post",
+      "content": "This is the first post",
+      "author_id": "1"
+    }'``
+
+    Response:
+    ``{"message": "Post created"}``
+    """
+    try:
+        table.put_item(Item={k: v for k, v in post.model_dump().items()})
+
+        return {"message": "Post created"}
+    except ClientError as e:
+        raise HTTPException(status_code=500, detail=e.response["Error"]["Message"])
+
+
+@router.put("/{Id}")
+def update_post(Id: str, post: Post):
+    """
+    Updates a post in the DynamoDB table "posts" by Id
+
+    Example call:
+    ``curl -X PUT "http://127.0.0.1:8000/posts/1" -H "accept: application/json" -H "Content-Type: application/json" -d  # noqa E501
+    '{
+        "Id": "1",
+        "title": "Updated Post",
+        "content": "This is the updated post",
+        "author_id": "1"
+    }'
+
+    Response:
+    ``{"message": "Post with id {Id} updated", "updated_attributes": {updated_attributes}}``
+    """
+    try:
+        response = table.update_item(
+            Key={"Id": Id},
+            UpdateExpression="set title=:t, content=:c, author_id=:a",
+            ExpressionAttributeValues={":t": post.title, ":c": post.content, ":a": post.author_id},
+            ReturnValues="UPDATED_NEW",
+        )
+        return {
+            "message": f"Post with id {Id} updated",
+            "updated_attributes": response["Attributes"],
+        }
+    except ClientError as e:
+        raise HTTPException(status_code=500, detail=e.response["Error"]["Message"])
+
+
+@router.delete("/{Id}")
+def delete_post(Id: str):
+    """
+    Deletes a post from the DynamoDB table "posts" by Id
+
+    Example call:
+    ``curl -X DELETE "http://127.0.0.1:8000/posts/1" -H "accept: application/json"``
+
+    Response:
+    ``{"message":"Post with id 1 deleted"}
+    """
+    try:
+        table.delete_item(Key={"Id": Id})
+        return {"message": f"Post with id {Id} deleted"}
+    except ClientError as e:
+        raise HTTPException(status_code=500, detail=e.response["Error"]["Message"])

app/routes/__init__.py

@@ -0,0 +1,132 @@
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
+import boto3
+from botocore.exceptions import ClientError
+
+router = APIRouter()
+
+# Initialize the DynamoDB client to connect to AWS DynamoDB
+dynamodb = boto3.resource("dynamodb")
+table = dynamodb.Table("users")
+
+
+class User(BaseModel):
+    Id: str  # Change user_id to Id to match the DynamoDB table schema
+    user_name: str
+    password: str
+    email_address: str
+
+
+@router.post("/")
+def create_user(user: User):
+    """
+    Creates a new user in the DynamoDB table "users"
+
+    Example call:
+    ``curl -X POST "http://127.0.0.1:8000/users/" -H "accept: application/json" -H "Content-Type: application/json" -d '{  # noqa E501
+      "Id": "1",
+      "user_name": "johndoe",
+      "password": "password",
+      "email_address": "[email protected]"
+    }'``
+
+    Response: ``{"message":"User created"}``
+    """
+    try:
+        table.put_item(
+            Item={
+                "Id": user.Id,
+                "user_name": user.user_name,
+                "password": user.password,
+                "email_address": user.email_address,
+            }
+        )
+        return {"message": "User created"}
+    except ClientError as e:
+        raise HTTPException(status_code=500, detail=e.response["Error"]["Message"])
+
+
+@router.get("/{Id}")
+def read_user(Id: str):
+    """
+    Reads a user from the DynamoDB table "users" by Id
+
+    Example call:
+    ``curl -X GET "http://127.0.0.1:8000/users/1" -H "accept: application/json"``
+
+    Response: ``{"Id": "1", "user_name": "johndoe", "password": ...``
+    """
+    try:
+        response = table.get_item(Key={"Id": Id})
+        if "Item" not in response:
+            raise HTTPException(status_code=404, detail="User not found")
+        return response["Item"]
+    except ClientError as e:
+        raise HTTPException(status_code=500, detail=e.response["Error"]["Message"])
+
+
+@router.get("/")
+def read_users():
+    """
+    Reads all users from the DynamoDB table "users"
+
+    Example call:
+    ``curl -X GET "http://127.0.0.1:8000/users/" -H "accept: application/json"``
+
+    Response: ``[{"Id": "1", "user_name": "johndoe", "password": "password", "email_address": ...``
+    """
+    try:
+        response = table.scan()
+        return response["Items"]
+    except ClientError as e:
+        raise HTTPException(status_code=500, detail=e.response["Error"]["Message"])
+
+
+@router.put("/{Id}")
+def update_user(Id: str, user: User):
+    """
+    Updates a user in the DynamoDB table "users" by Id
+
+    Example call:
+    ``curl -X PUT "http://127.0.0.1:8000/users/{Id}" -H "accept: application/json" -H "Content-Type: application/json" -d '{  # noqa E501
+      "Id": "1",
+      "user_name": "johndoeupdated",
+      "password": "newpassword",
+      "email_address": "[email protected]"
+    }'
+
+    Response: ``{"message":"User with id 1 updated","updated_attributes": ...}``
+    """
+    try:
+        response = table.update_item(
+            Key={"Id": Id},
+            UpdateExpression="set user_name=:u, password=:p, email_address=:e",
+            ExpressionAttributeValues={
+                ":u": user.user_name,
+                ":p": user.password,
+                ":e": user.email_address,
+            },
+            ReturnValues="UPDATED_NEW",
+        )
+        return {
+            "message": f"User with id {Id} updated",
+            "updated_attributes": response["Attributes"],
+        }
+    except ClientError as e:
+        raise HTTPException(status_code=500, detail=e.response["Error"]["Message"])
+
+
+@router.delete("/{Id}")
+def delete_user(Id: str):
+    """
+    Deletes a user from the DynamoDB table "users" by Id
+
+    Example call: ``curl -X DELETE "http://127.0.0.1:8000/users/1" -H "accept: application/json"``
+
+    Response: ``{"message":"User with id 1 deleted"}``
+    """
+    try:
+        table.delete_item(Key={"Id": Id})
+        return {"message": f"User with id {Id} deleted"}
+    except ClientError as e:
+        raise HTTPException(status_code=500, detail=e.response["Error"]["Message"])

app/routes/posts.py

@@ -0,0 +1,28 @@
+import json
+import logging
+from app.main import handler
+
+logging.basicConfig(level=logging.INFO)
+
+
+def lambda_handler(event, context):
+    """
+    Lambda handler function
+
+    Args:
+        event (dict): Event data
+        context (object): Runtime information
+    """
+    logging.info(f"Received event: {json.dumps(event, indent=2)}")
+
+    try:
+        response = handler(event, context)
+    except Exception as e:
+        logging.error(f"Error: {e}")
+        response = {
+            "statusCode": 500,
+            "headers": {"Content-Type": "application/json"},
+            "body": json.dumps({"error": str(e)}),
+        }
+
+    return response

app/routes/users.py

@@ -3,8 +3,11 @@ pytest-asyncio
 pytest-cov
 
 boto3 == 1.35.72
+fastapi == 0.115.*
+httpx == 0.28.*
 pydantic == 2.9.*
-datamodel-code-generator == 0.26.*
+mangum == 0.19.*
+uvicorn == 0.32.*