rogulski.it

My name is Piotr, a passionate pythonista and this is my blog!

Using DynamoDB with FastAPI

Posted at — Jan 22, 2022

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.

Plan

In this blog post we will:

  1. First, setup DynamoDB engine on the local machine
  2. Write some code to connect DynamoDB with FastAPI
  3. Test our code with pytest
  4. Create terraform file to create our table on AWS side

Local setup

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

Project setup

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.

DynamoDB definition in terraform

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"
  }
}

DynamoDB definition in Python code

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.

DynamoDB table usage

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)

FastAPI endpoints

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'

Summary

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