Building Scalable Serverless CRUD APIs with FastAPI and AWS DynamoDB Local: A Step-by-Step Guide

Revolutionize the way you develop applications by seamlessly integrating Python FastAPI and the power of AWS DynamoDB.

DynamoDB: AWS’s Managed NoSQL Powerhouse

DynamoDB, an AWS’s Managed NoSQL Powerhouse DynamoDB, an AWS-managed NoSQL database, boasts effortless scalability, low-latency access, and automatic replication. Its prowess lies in effectively managing substantial data loads and high-speed read/write operations, making it an ideal solution for diverse data handling needs.

FastAPI: Speed and Simplicity in Python FastAPI

FastAPI, a contemporary Python web framework, excels in rapid API development with its agility and ease. It handles vital tasks such as data validation, serialization, and documentation, empowering developers to effortlessly craft robust APIs. With asynchronous programming support, FastAPI is tailor-made for real-time applications.

FastAPI vs. Traditional REST APIs:

Speed and Efficiency:

• FastAPI: Utilizes asynchronous programming for high performance, handling more requests with fewer resources.
• REST APIs: Generally synchronous, which can lead to slower response times under heavy loads.

Type Safety and Validation:

• FastAPI: Leverages Python type hints for automatic data validation and interactive API documentation.
• REST APIs: Often require manual validation and lack built-in interactive documentation.

Framework Features:

• FastAPI: Modern framework with automatic serialization, dependency injection, and native WebSocket support.
• REST APIs: May need additional libraries for similar features, potentially adding complexity.

Development Speed:

• FastAPI: Quick development with automatic endpoint generation and reduced boilerplate code.
• REST APIs: More manual setup and configuration, possibly leading to longer development cycles.

Why Combine FastAPI and DynamoDB?

By combining FastAPI with DynamoDB, you leverage the strengths of both. FastAPI’s efficient API design and asynchronous capabilities align perfectly with DynamoDB’s scalability and low-latency access. This synergy enables developers to build high-performance APIs powered by a dynamic, robust database, enhancing user experiences significantly.

Local DynamoDB Configuration:

We will begin the process by setting up our DynamoDB environment through the straightforward method of downloading Docker. Download it from here https://docs.docker.com/engine/install/

Run below command to run DynamoDB locally, Create a folder structure. This is how the folder structure should look like:

There are two ways to start docker, either we can start it globally or you can go to docker-compose.yml, start a terminal. Open a terminal or command prompt.

1.Pull the DynamoDB Local Docker image:

docker pull amazon/dynamodb-local

2.Run the DynamoDB Local container:

docker run -p 8000:8000 amazon/dynamodb-local

Open a new terminal window in the same project folder, if you are running it from the terminal. This terminal will be used for additional commands without interrupting the running DynamoDB container.

Dynamodb Admin:

Setting up DynamoDB Admin (a web-based GUI tool for managing your DynamoDB tables) can be helpful for visualizing and managing your local DynamoDB instance. Here’s how you can set up DynamoDB Admin to work with DynamoDB Local:

Go to FastAPI, install DynamoDB Admin using npm (Node.js package manager). Make sure you have Node.js installed on your machine.

npm install -g dynamodb-admin

Run DynamoDB Admin, specifying the local DynamoDB endpoint.

DYNAMO_ENDPOINT=http://localhost:8000

Make sure to replace http://localhost:8000 with the appropriate endpoint if you are running DynamoDB Local on a different host or port.

You can go to the browser, run http://localhost:8001/ and access the web based UI, since the aim of this article is to integrate fastapi with dynamodb, we won’t go deep into it but you can create tables as it happens in aws.Now, let’s create a virtual environment for running before we get started with fastApi

Creating a virtual environment is a best practice in Python development. It isolates project-specific dependencies, ensuring that different projects can have their own distinct set of libraries without interfering with each other. This will be the command for creating virtual environment

python -m venv venv

Enter the Below command in terminal this will activate venv:

.\venv\Scripts\activate

Install dependencies

pip install fastapi[all]

Dependencies Explanation:

pip install: This is the same command used to install Python packages.

fastapi[all]: This uses an “extra” called all provided by the FastAPI package. An “extra” is a way to install a package along with additional optional dependencies.

pip install uvicorn boto3

Dependencies Explanation:

uvicorn: This is an ASGI server implementation that is commonly used to run FastAPI applications. It provides high performance and supports asynchronous programming.

boto3: This is the Amazon Web Services (AWS) SDK for Python. It allows you to interact with various AWS services, including DynamoDB (a NoSQL database service).

Docker compose configuration:

Configuration via docker-compose.yml

version: "3"
services:
my_app:
build:
context: .
dockerfile: Dockerfile
ports:
- "8000:80"
volumes:
- ./app:/app

This Docker Compose configuration defines a service named “application.” It creates an image based on the Dockerfile in the current directory, forwards traffic from port 8000 on the host to port 80 in the container, and establishes a volume link between the local “app” directory and the “/app” directory inside the container.

DockerFile content

FROM tiangolo/uvicorn-gunicorn-fastapi:python3.8
COPY ./app /app

This Dockerfile starts with a base image named “tiangolo/uvicorn-gunicorn-fastapi” that includes Python 3.8 and the Uvicorn ASGI server. The “COPY” command is used to copy the contents of the local “./app” directory to the “/app” directory inside the container. This establishes the application code’s presence within the container, making it ready to be executed by the specified server.

# main.py

from fastapi import FastAPI

app = FastAPI()
@app.get("/")
def read_root():
    return {"Hello": "World"}

To confirm proper FastAPI configuration, execute the following command:

uvicorn main:app - host 0.0.0.0 - port 8000

Go to the URL, if you see {“Hello”: “World”}, then congratulations, you’ve just configured FastAPI successfully.

To begin our CRUD operations, the initial step involves table creation. Let’s establish a “todo” table as the foundation.

Create a createTable.py file in dynamodb folder

import boto3
dynamodb = boto3.resource('dynamodb', endpoint_url='http://localhost:8000')
try:
    table = dynamodb.create_table(
    TableName='TodoTable',
    KeySchema=[
    {
    'AttributeName': 'itemId',
    'KeyType': 'HASH'
    }
    ],
    AttributeDefinitions=[
    {
    'AttributeName': 'itemId',
    'AttributeType': 'S'
    }
    ],
    ProvisionedThroughput={
    'ReadCapacityUnits': 5,
    'WriteCapacityUnits': 5
    }
    )
    print("Table status:", table.table_status)
except Exception as e:
print('error', e)

Now try to run the command python createTable.py, if you get “Table status:Active” in the terminal, then table creation is successful. Functions for crud Api in crud.py

import boto3
from botocore.exceptions import ClientError
from fastapi import HTTPException

dynamodb = boto3.resource('dynamodb', endpoint_url='http://localhost:8000')
table = dynamodb.Table('TodoTable')
def create_item(item):
    existing_item = get_item(item['itemId'])
    if existing_item:
         raise HTTPException(status_code=400, detail="Item already exists")
    try:
        response = table.put_item(Item=item)
        return item # Return the created item itself
    except ClientError as e:
        raise HTTPException(status_code=500, detail=str(e))
def get_items():
    response = table.scan()
    return response.get('Items', [])
def get_item(item_id):
    response = table.get_item(Key={'itemId': item_id})
    return response.get('Item', None)
    
def update_item(item_id, new_attributes):
# Check if the item exists before attempting to update
    existing_item = get_item(item_id)
    if not existing_item:
        raise HTTPException(status_code=400, detail="No item found")
    try:
        response = table.update_item(
        Key={'itemId': item_id},
        UpdateExpression='SET title = :title, description = :description, done = :done',
        ExpressionAttributeValues={
        ':title': new_attributes.get('title'),
        ':description': new_attributes.get('description'),
        ':done': new_attributes.get('done')
        },
        ReturnValues='ALL_NEW'
        )
        return response.get('Attributes', {})
    except ClientError as e:
        raise HTTPException(status_code=500, detail=str(e))
def delete_item(item_id):
    try:
        response = table.delete_item(Key={'itemId': item_id})
    except ClientError as e:
        raise HTTPException(status_code=500, detail=str(e))
return response

Now lets create routers in main.py:

from typing import List
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from app.crud import create_item, get_items, get_item, update_item, delete_item

app = FastAPI()
class TodoItem(BaseModel):
itemId: str
title: str
description: str
done: bool = False
@app.get("/")
def read_root():
    return {"message": "Welcome to the FastAPI DynamoDB Todo App!"}
@app.get("/items/", response_model=List[TodoItem])
def read_todo_items():
    return get_items()
@app.post("/items", response_model=TodoItem)
def create_todo_item(item: TodoItem):
    created_item = create_item(item.model_dump())
    return created_item # Ensure that create_item returns the item itself
@app.get("/items/{item_id}", response_model=TodoItem)
def read_todo_item(item_id: str):
    item = get_item(item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item
@app.put("/items/{item_id}", response_model=TodoItem)
def update_todo_item(item_id: str, updates: TodoItem):
    item = update_item(item_id, updates.model_dump())
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item
@app.delete("/items/{item_id}", response_model=dict)
def delete_todo_item(item_id: str):
    response = delete_item(item_id)
    if 'ResponseMetadata' not in response:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"message": "Item deleted successfully"}

You have the option to test it either using Postman or a UI-based framework for FastAPI. To test it within the FastAPI framework, simply run the following URL: “localhost:8080/docs”. Additionally, there’s another URL option available: “localhost:8080/redoc”.

About the Author:

Poonam Tomar is a budding Software Engineer with a year of hands-on experience in the realm of software development. Throughout her journey, she has delved into diverse projects, displaying a keen aptitude for unraveling intricate challenges. Her passion for continuous learning shines through as she embraces various facets of the field, including problem-solving, web development, and an insightful grasp of architectural analysis.

About CodeStax.Ai

At CodeStax.Ai, we stand at the nexus of innovation and enterprise solutions, offering technology partnerships that empower businesses to drive efficiency, innovation, and growth, harnessing the transformative power of no-code platforms and advanced AI integrations.

But the real magic? It’s our tech tribe behind the scenes. If you’ve got a knack for innovation and a passion for redefining the norm, we’ve got the perfect tech playground for you. CodeStax.Ai offers more than a job — it’s a journey into the very heart of what’s next. Join us, and be part of the revolution that’s redefining the enterprise tech landscape.