Matia Rašetina

Posted on Nov 25

PynamoDB Tutorial: Build Production-Ready DynamoDB CRUD APIs in AWS Lambda

#aws #database #serverless #programming

I've been building on AWS for multiple years now, and one of the most repetitive parts I found was writing crude operations for DynamoDB. Every project starts the same:

use boto3 to define the table
define a big JSON payload JUST to fetch some data from the table
copy-and-paste methods from previous projects for all CRUD operations, basically adding boilerplate code

That's when I found out about a library called PynamoDB , a lightweight ORM for DynamoDB tables and that changed the way I manipulate table’s data.

Welcome to the part two of creating code for a real estate platform where in the first post we went over a pipeline of labeling main parts of the properties and now we're gonna go over Lambdas which are responsible for CRUD operations.

The code in this post will be a shortened version than the one from the full project, just to make it as simple as possible. The full code used on the platform will be available in the next blog posts.

In this post, I’ll break down how I put together a simple CRUD stack using:

PynamoDB for the data model
Lambda functions for the business logic and interaction with the DynamoDB table via PynamoDB model

The whole goal is ease of use — a DynamoDB-backed CRUD pattern that feels as natural as working with a traditional ORM, but with all the serverless benefits of AWS.

AWS CDK code won’t be mentioned in this blog post, however it will be available on the GitHub repository which you can find by clicking on this link.

Which AWS resource are we using?

All resources will be created via AWS CDK in Python.

We will start off with AWS Lambda - here is a short description of the Lambdas and what they do:

Create Property Lambda - enter received property information, put it into the DynamoDB table and provide the user pre-signed URLs for property image uploads to the S3 bucket
Get Property Lambda - read all or a specific property from the database
Update Property Lambda - update the specific property entry which is inside the database
Delete Property Lambda - delete all images and entry inside the database

Of course, since we are showcasing the PynamoDB library, we need to use DynamoDB as our database.

Why should I use PynamoDB instead of boto3.client?

Well, great question! Take a look at the Python snippets which compares the syntax for both libraries:

With boto3.client:

item = {
    'PK': {'S': f'PROPERTY#{property_id}'},
    'price': {'N': '500000'},
    'title': {'S': 'Beautiful Home'},
    'createdAt': {'S': '2024-01-15T10:30:00Z'}
}
dynamodb.put_item(TableName='Properties', Item=item)

With PynamoDB:

property_item = PropertyModel(
    property_id=property_id,
    price=500000,
    title='Beautiful Home',
    createdAt=datetime.now()
)
property_item.save()

Notice how PynamoDB handles type conversion automatically - no more wrapping everything in {'S': ...} or {'N': ...} dictionaries!

Designing the PynamoDB model

Here is the PynamoDB model which I’ve used for my real estate platform:

from datetime import datetime, timezone
from os import environ
from pynamodb.models import Model
from pynamodb.attributes import (
    UnicodeAttribute, NumberAttribute, UTCDateTimeAttribute, MapAttribute,
    ListAttribute, BooleanAttribute, JSONAttribute
)

class PropertyModel(Model):
    class Meta:
        table_name = environ.get('PROPERTY_TABLE_NAME', 'SimplifiedPropertyTable')
        region = environ.get('AWS_REGION', 'eu-central-1')

    PK = UnicodeAttribute(hash_key=True)
    SK = UnicodeAttribute(range_key=True)

    id = UnicodeAttribute()
    title = UnicodeAttribute(null=True)
    description = UnicodeAttribute(null=True)
    price = NumberAttribute(null=True)
    address = UnicodeAttribute(null=True)
    city = UnicodeAttribute(null=True)
    state = UnicodeAttribute(null=True)
    zip = UnicodeAttribute(null=True)

    createdAt = UTCDateTimeAttribute(null=True)
    updatedAt = UTCDateTimeAttribute(null=True)

    def __init__(self, property_id: str = None, **kwargs):
        now = datetime.now(timezone.utc)
        if property_id:
            kwargs.setdefault('PK', f"PROPERTY#{property_id}")
            kwargs.setdefault('SK', "METADATA")
            kwargs.setdefault('id', property_id)
            kwargs.setdefault('createdAt', now)
            kwargs.setdefault('updatedAt', now)
        super().__init__(**kwargs)

You can see that there are many information here, from numbers, booleans, lists, strings.

When using PynamoDB, it’s important to understand how PynamoDB interprets your data. You can see many new objects like UnicodeAttribute , BooleanAttribute , ListAttribute and others. Here are the Python-PynamoDB equivalents:

NumberAttribute - integer or float
ListAttribute - list or array
UnicodeAttribute - string
BooleanAttribute - boolean
JSONAttribute - a JSON object
MapAttribute - Python’s dictionary
UTCDateTimeAttribute - Python’s datetime.now() method

In this case, every single information about the property was put inside only one database. You may ask “Why?”

Two reasons which were crucial for this project:

makes the usage of DynamoDB much cheaper — if we had 2 or more tables, to fetch the data we’d need to call multiple tables, but this way we only call one, which reduces cost
no schema migrations - any new data can be easily added to the already existing DynamoDB entry. As I wanted to complete this project as soon as possible, I kept adding data to the table. DynamoDB enabled me that without complaining once.

Also, do you notice the Meta class?

The Meta class is incredibly powerful because it externalizes your configuration. Here's what's happening:

class Meta:
    table_name = environ.get('PROPERTY_TABLE_NAME', 'SimplifiedPropertyTable')
    region = environ.get('AWS_REGION', 'eu-central-1')

This means you can use the same code across environments:

Development:

export PROPERTY_TABLE_NAME=dev-properties-table
export AWS_REGION=us-east-1

Production:

export PROPERTY_TABLE_NAME=prod-properties-table
export AWS_REGION=eu-central-1

No code changes needed - just different environment variables. This is especially useful when deploying with CDK, as you can set these per stack/stage automatically.

CRUD operations with the PynamoDB model

Create DynamoDB Entry Lambda

Here is a very simple method which leverages the created PynamoDB model and saves the property information with ease, while creating pre-signed URLs for property images - just use the .save() method:

def handler(event, context):
    """
    Create a new property and return presigned URLs for image uploads
    """
    try:
        # Parse request body
        body = json.loads(event.get('body', '{}'))

        # Validate required fields
        required_fields = ['title', 'description', 'price', 'address', 'city', 'state', 'zip']
        missing_fields = [field for field in required_fields if field not in body]

        if missing_fields:
            logger.warning("Missing required fields", extra={"missing": missing_fields})
            return {
                "statusCode": 400,
                "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
                "body": json.dumps({"error": f"Missing required fields: {', '.join(missing_fields)}"})
            }

        # Generate unique property ID
        property_id = str(uuid.uuid4())

        # Create property in DynamoDB
        property_item = PropertyModel(
            property_id=property_id,
            title=body['title'],
            description=body['description'],
            price=body['price'],
            address=body['address'],
            city=body['city'],
            state=body['state'],
            zip=body['zip']
        )
        property_item.save()

        logger.info("Property created", extra={"property_id": property_id})

        return {
            "statusCode": 201,
            "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
            "body": json.dumps({
                "property_id": property_id,
                "property": property_item.to_dict()
            })
        }

    except Exception as e:
        logger.exception("Error creating property")
        return {
            "statusCode": 500,
            "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
            "body": json.dumps({"error": "Failed to create property"})
        }

Get DynamoDB Entry Lambda

There are 2 flows for this Lambda:

if the property ID is provided in the request, query the database for information just for that property
scan the table and get all the data

To scan the database, use the pynamodb_model.scan() method or to fetch the precise item from the database, use the pynamodb_model.get('ID_of_item_you_want_to_fetch'):

def handler(event, context):
    """
    Get a single property by ID or list all properties
    """
    try:
        # Check if property_id is provided in query parameters
        query_params = event.get('queryStringParameters') or {}
        property_id = query_params.get('propertyId')

        if property_id:
            # Get single property and handle the case
            # if the property requested doesn't exist
            try:
                property_item = PropertyModel.get(
                    hash_key=f"PROPERTY#{property_id}",
                    range_key="METADATA"
                )
            except PropertyModel.DoesNotExist:
                property_item = None

            if not property_item:
                logger.warning("Property not found", extra={"property_id": property_id})
                return {
                    "statusCode": 404,
                    "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
                    "body": json.dumps({"error": "Property not found"})
                }

            logger.info("Property retrieved", extra={"property_id": property_id})
            return {
                "statusCode": 200,
                "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
                "body": json.dumps({"property": property_item.to_dict()})
            }

        else:
            # List all properties (scan operation)
            properties = []
            for item in PropertyModel.scan():
                properties.append(item.to_dict())

            logger.info("Properties listed", extra={"count": len(properties)})
            return {
                "statusCode": 200,
                "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
                "body": json.dumps({
                    "properties": properties,
                    "total": len(properties)
                })
            }

    except Exception as e:
        logger.exception("Error retrieving property")
        return {
            "statusCode": 500,
            "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
            "body": json.dumps({"error": "Failed to retrieve property"})
        }

Important note on scanning: The PropertyModel.scan() operation shown above works great for small datasets, but scanning becomes expensive and slow as your table grows. In production, consider:

Pagination: For large result sets, use last_evaluated_key to paginate:

results = PropertyModel.scan(limit=20)
for item in results:
    properties.append(item.to_dict())

# Get the last evaluated key for next page
if results.last_evaluated_key:
    next_results = PropertyModel.scan(
        limit=20,
        last_evaluated_key=results.last_evaluated_key
    )

Query instead of Scan: If you frequently filter by city or state, add a Global Secondary Index (GSI) and use PropertyModel.city_index.query('Vienna') instead of scanning everything.
Limit for safety: Always set a limit in production to prevent timeout issues: PropertyModel.scan(limit=100)

For my real estate platform with hundreds of properties, I added pagination and GSIs for common queries - I'll cover this in a future post!

Update DynamoDB Entry Lambda

Here is the code for updating the property already entered in the platform by using pynamodb_model.update() method:

def handler(event, context):
    """
    Update an existing property
    """
    try:
        # Get property ID from path parameters
        property_id = event.get('pathParameters', {}).get('id')

        if not property_id:
            return {
                "statusCode": 400,
                "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
                "body": json.dumps({"error": "Property ID is required"})
            }

        # Parse request body
        body = json.loads(event.get('body', '{}'))

        # Get existing property
        try:
            property_item = PropertyModel.get(
                hash_key=f"PROPERTY#{property_id}",
                range_key="METADATA"
            )
        except PropertyModel.DoesNotExist:
            property_item = None

        if not property_item:
            logger.warning("Property not found", extra={"property_id": property_id})
            return {
                "statusCode": 404,
                "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
                "body": json.dumps({"error": "Property not found"})
            }

        # Update fields
        update_actions = []
        updatable_fields = ['title', 'description', 'price', 'address', 'city', 'state', 'zip']

        for field in updatable_fields:
            if field in body:
                update_actions.append(
                    getattr(PropertyModel, field).set(body[field])
                )

        # Always update the updatedAt timestamp
        update_actions.append(
            PropertyModel.updatedAt.set(datetime.now())
        )

        if update_actions:
            property_item.update(actions=update_actions)

        # Refresh the item to get updated values
        property_item.refresh()

        logger.info("Property updated", extra={"property_id": property_id})
        return {
            "statusCode": 200,
            "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
            "body": json.dumps({"property": property_item.to_dict()})
        }

    except Exception as e:
        logger.exception("Error updating property")
        return {
            "statusCode": 500,
            "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
            "body": json.dumps({"error": "Failed to update property"})
        }

Delete DynamoDB Entry Lambda

The request should provide the property ID, and delete the property details by using the pynamodb_model.delete() method:

def handler(event, context):
    """
    Delete a property by ID
    """
    try:
        # Get property ID from path parameters
        property_id = event.get('pathParameters', {}).get('id')

        if not property_id:
            return {
                "statusCode": 400,
                "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
                "body": json.dumps({"error": "Property ID is required"})
            }

        # Get existing property
        # and catch the PropertyModel.DoesNotExist error 
        # if the property ID is not found inside the DynamoDB
        try:
            property_item = PropertyModel.get(
                hash_key=f"PROPERTY#{property_id}",
                range_key="METADATA"
            )
        except PropertyModel.DoesNotExist:
            property_item = None

        if not property_item:
            logger.warning("Property not found", extra={"property_id": property_id})
            return {
                "statusCode": 404,
                "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
                "body": json.dumps({"error": "Property not found"})
            }

        # Delete the property
        property_item.delete()

        logger.info("Property deleted", extra={"property_id": property_id})
        return {
            "statusCode": 200,
            "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
            "body": json.dumps({"message": "Property deleted successfully"})
        }

    except Exception as e:
        logger.exception("Error deleting property")
        return {
            "statusCode": 500,
            "headers": {'Access-Control-Allow-Origin': ALLOWED_ORIGIN},
            "body": json.dumps({"error": "Failed to delete property"})
        }

Error and Race Condition Handling

One thing to be aware of when building production CRUD operations: concurrent updates.

Imagine two users trying to update the same property price simultaneously. Without proper handling, the last write wins and one update gets lost.

PynamoDB supports conditional writes using conditions:

from pynamodb.expressions.condition import Condition

def handler(event, context):
    # ... existing code to get property_item ...

    # Only update if the price hasn't changed since we read it
    original_price = property_item.price

    try:
        property_item.update(
            actions=[PropertyModel.price.set(new_price)],
            condition=(PropertyModel.price == original_price)
        )
    except PropertyModel.UpdateError:
        return {
            "statusCode": 409,
            "body": json.dumps({
                "error": "Property was modified by another user. Please refresh and try again."
            })
        }

This is especially important for sensitive fields like price or availability status. For my platform, I use conditional updates when changing property status to prevent double-booking scenarios.

Deployment process

Very easy! You can take the GitHub repository provided and just run cdk deploy inside your terminal window!

Just make sure your AWS credentials are correctly setup on your machine.

Future work & Conclusion

Now that we went over how to use PynamoDB ORM usage, the next questions should be:

how do we want to protect our CreateProperty , UpdateProperty and DeleteProperty Lambdas?
- my choice? AWS Cognito makes it very easy to protect the endpoints via an AWS API Gateway Authorizer method
single Lambda handling multiple HTTP methods?
- everything is in one place, however code maintenance might become an issue
Improving Lambda performance by implementing Global Secondary Indexes inside the DynamoDB table
- this could be implemented based on which data the users mostly get

Ever since I’ve found out about PynamoDB, I’ve been using it constantly in every project. It just makes it so easy to add, access and modify the data which is in the DynamoDB table. Hopefully it helps you in your coding!

Thank you for reading! See you in the next post!

DEV Community