DynamoDB is a NOSQL database from AWS, with a couple of years of experience it features a lot of amazing functionalities. It is way more than “just a key-value storage”. It can hold relations between multiple records, scale efficiently and what is most important it is super fun to use!
In this post, I would like to show how simple is to start using DynamoDB with Python, FastAPI. This example assumes you already know how DynamoDB works, especially Tables and Global Secondary Indexes.
In this blog post we will:
We will create a docker-compose file, using two docker images. First amazon/dynamodb-local to locally develop and test our connection. Second aaronshaf/dynamodb-admin to be able to see our tables in the web view on the page.
version: "3.7"
services:
dynamodb:
image: amazon/dynamodb-local
expose:
- 8000
ports:
- 8000:8000
dynamodb-admin:
image: aaronshaf/dynamodb-admin
environment:
DYNAMO_ENDPOINT: http://dynamodb:8000
ports:
- 8001:80011
Our project structure will look like this:
├── Dockerfile
├── LICENSE
├── README.md
├── app
│ ├── __init__.py
│ ├── main.py
│ ├── repositories.py
│ ├── schemas.py
│ ├── settings.py
│ ├── tables.py
│ └── tests
│ ├── __init__.py
│ ├── conftest.py
│ └── test_main.py
├── docker-compose.yaml
├── docs
│ └── swagger-docs.png
├── poetry.lock
├── pyproject.toml
└── tf
└── resource_dynamodb.tf
Only for example purposes, we will not implement a base abstract classes, post will be simple and easy. Feel free to modify it as you wish.
To keep our table consistent and reliable we will use terraform config to set up the table. It should include only definitions of columns which are hash or range keys, both for table and Global Secondary Index.
resource "aws_dynamodb_table" "product-table" {
name = "ProductTable"
billing_mode = "PROVISIONED"
read_capacity = 20
write_capacity = 20
hash_key = "id"
attribute {
name = "id"
type = "S"
}
attribute {
name = "name"
type = "N"
}
attribute {
name = "updated_at"
type = "N"
}
ttl {
attribute_name = "TimeToExist"
enabled = false
}
global_secondary_index {
name = "product-name-index"
hash_key = "name"
range_key = "created_at"
write_capacity = 10
read_capacity = 10
projection_type = "INCLUDE"
non_key_attributes = ["id"]
}
tags = {
Name = "product-table"
Environment = "production"
}
}
To represent tables and use them in the code we will use pynamodb. Unfortunately, at the time this guide is created, they are not supporting async yet. If async is very important for you, please check boto3 support for DynamoDB.
from pynamodb.attributes import NumberAttribute, UnicodeAttribute
from pynamodb.indexes import AllProjection, GlobalSecondaryIndex
from pynamodb.models import Model
from app.settings import config
class BaseTable(Model):
class Meta:
host = config.DB_HOST if config.ENVIRONMENT in ["local", "test"] else None
region = config.AWS_REGION
class ProductNameIndex(GlobalSecondaryIndex["ProductTable"]):
"""
Represents a global secondary index for ProductTable
"""
class Meta:
index_name = "product-name-index"
read_capacity_units = 10
write_capacity_units = 10
projection = AllProjection()
name = UnicodeAttribute(hash_key=True)
updated_at = NumberAttribute(range_key=True)
class ProductTable(BaseTable):
"""
Represents a DynamoDB table for a Product
"""
class Meta(BaseTable.Meta):
table_name = "product-table"
id = UnicodeAttribute(hash_key=True)
name = UnicodeAttribute(null=False)
description = UnicodeAttribute(null=False)
created_at = NumberAttribute(null=False)
updated_at = NumberAttribute(null=False)
product_name_index = ProductNameIndex()
As a hash_key to the table, we are using UUID, range_key is not specified. Next, to be able to efficiently query thru the product name with updated_at order we are setting up Global Secondary Index.
To keep everything in the same place we will use Repository Pattern and put all DB operations needed for this tutorial.
import time
import uuid
from typing import Dict, Any, Union
from app.schemas import ProductSchemaIn, ProductSchemaOut
from app.tables import ProductTable
class ProductRepository:
table: ProductTable = ProductTable
schema_out: ProductSchemaOut = ProductSchemaOut
@staticmethod
def _preprocess_create(values: Dict[str, Any]) -> Dict[str, Any]:
timestamp_now = time.time()
values["id"] = str(uuid.uuid4())
values["created_at"] = timestamp_now
values["updated_at"] = timestamp_now
return values
@classmethod
def create(cls, product_in: ProductSchemaIn) -> ProductSchemaOut:
data = cls._preprocess_create(product_in.dict())
model = cls.table(**data)
model.save()
return cls.schema_out(**model.attribute_values)
@classmethod
def get(cls, entry_id: Union[str, uuid.UUID]) -> ProductSchemaOut:
model = cls.table.get(str(entry_id))
return cls.schema_out(**model.attribute_values)
Finally, let’s allow us to create and get our products from the DynamoDB table using FastAPI.
app = FastAPI()
@app.post(
"/v1/products",
status_code=status.HTTP_201_CREATED,
response_model=ProductSchemaOut,
)
def create_product(product_in: ProductSchemaIn) -> ProductSchemaOut:
product_out = ProductRepository.create(product_in)
return product_out
@app.get(
"/v1/products/{product_id}",
status_code=status.HTTP_200_OK,
response_model=ProductSchemaOut,
)
def create_product(product_id: UUID) -> ProductSchemaOut:
product_out = ProductRepository.get(product_id)
return product_out
Run your project using docker compose:
docker compose up -d
Create product:
curl -X 'POST' \
'http://0.0.0.0:8080/v1/products' \
-H 'accept: application/json' \
-H 'Content-Type: application/json' \
-d '{
"name": "Book",
"description": "Another paper thing"
}'
Get product:
curl -X 'GET' \
'http://0.0.0.0:8080/v1/products/3fa85f64-5717-4562-b3fc-2c963f66afa6' \
-H 'accept: application/json'
I hope this code sample will help you build amazing projects. I have already run this on production and it is working like a charm. Of course, it is not 1:1 what I use but the stack and most of the configs are the same: DynamoDB and FastAPI.
Free free to check the full source code of this project on my GitHub