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.


    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"
        image: amazon/dynamodb-local
          - 8000
          - 8000:8000
        image: aaronshaf/dynamodb-admin
          DYNAMO_ENDPOINT: http://dynamodb:8000
          - 8001:80011

    Project setup

    Our project structure will look like this:

    ├── Dockerfile
    ├── LICENSE
    ├── app
    │   ├──
    │   ├──
    │   ├──
    │   ├──
    │   ├──
    │   ├──
    │   └── tests
    │       ├──
    │       ├──
    │       └──
    ├── docker-compose.yaml
    ├── docs
    │   └── swagger-docs.png
    ├── poetry.lock
    ├── pyproject.toml
    └── 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
        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
        def create(cls, product_in: ProductSchemaIn) -> ProductSchemaOut:
            data = cls._preprocess_create(product_in.dict())
            model = cls.table(**data)
            return cls.schema_out(**model.attribute_values)
        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()
    def create_product(product_in: ProductSchemaIn) -> ProductSchemaOut:
        product_out = ProductRepository.create(product_in)
        return product_out
    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' \
      '' \
      -H 'accept: application/json' \
      -H 'Content-Type: application/json' \
      -d '{
      "name": "Book",
      "description": "Another paper thing"

    Get product:

    curl -X 'GET' \
      '' \
      -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