Single-table design

Single-table design is a DynamoDB pattern where you store multiple entity types in one table. Instead of having separate tables for Users, Orders, and Products, you put them all in the same table and use key prefixes to distinguish them.

Why single-table?

DynamoDB charges per table and has limits on how many tables you can create. More importantly, single-table design lets you fetch related data in one query. In a multi-table design, getting a user and their orders requires two separate API calls. With single-table, you can get both in one query.

The tradeoff is complexity. You need to carefully design your keys to support all your access patterns. pydynox makes this easier with template keys.

Learn more

Alex DeBrie's The What, Why, and When of Single-Table Design is the best introduction to this pattern.

The problem

Without templates, you manually build keys and can make mistakes:

class User(Model):
    model_config = ModelConfig(table="app")
    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)

class Order(Model):
    model_config = ModelConfig(table="app")
    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)

# Bug: Order with User's key pattern - no error!
Order(pk="USER#john@example.com", sk="PROFILE")

There's no validation. You can accidentally create an Order with a User's key pattern and corrupt your data.

Template keys

Add template to StringAttribute to define key patterns:

template_basic.py

"""Example: Basic template keys."""

import asyncio

from pydynox import Model, ModelConfig
from pydynox.attributes import StringAttribute


class User(Model):
    model_config = ModelConfig(table="app")

    pk = StringAttribute(partition_key=True, template="USER#{email}")
    sk = StringAttribute(sort_key=True, template="PROFILE")
    email = StringAttribute()
    name = StringAttribute()


async def main():
    # Keys are built automatically from the template
    user = User(email="john@example.com", name="John")
    print(user.pk)  # "USER#john@example.com"
    print(user.sk)  # "PROFILE"

    await user.save()

    # Get by built keys
    found = await User.get(pk="USER#john@example.com", sk="PROFILE")
    if found:
        print(found.name)  # "John"

    # Cleanup
    await user.delete()


if __name__ == "__main__":
    asyncio.run(main())

The template USER#{email} means: take the email attribute value and put it after USER#. The {email} part is a placeholder that gets replaced with the actual value.

Static templates like PROFILE have no placeholders - they're always the same value.

Multiple placeholders

Templates can have multiple placeholders:

template_multiple.py

"""Example: Multiple placeholders in template."""

import asyncio

from pydynox import Model, ModelConfig
from pydynox.attributes import StringAttribute


class Order(Model):
    model_config = ModelConfig(table="app")

    pk = StringAttribute(partition_key=True, template="USER#{user_id}")
    sk = StringAttribute(sort_key=True, template="ORDER#{order_id}#{date}")
    user_id = StringAttribute()
    order_id = StringAttribute()
    date = StringAttribute()
    status = StringAttribute()


async def main():
    order = Order(user_id="123", order_id="456", date="2024-01-15", status="pending")
    print(order.pk)  # "USER#123"
    print(order.sk)  # "ORDER#456#2024-01-15"

    await order.save()

    # Query using placeholder
    async for o in Order.query(user_id="123"):
        print(f"{o.order_id}: {o.status}")

    # Cleanup
    await order.delete()


if __name__ == "__main__":
    asyncio.run(main())

This is useful when you need to sort by multiple attributes. Here, orders are sorted by order_id first, then by date.

Querying with templates

Query using placeholder names instead of building keys manually:

# Without template - you build the key yourself
async for order in Order.query(partition_key="USER#123"):
    print(order)

# With template - pass the placeholder value
async for order in Order.query(user_id="123"):
    print(order)

Both work. The template version is cleaner and less error-prone. pydynox builds USER#123 for you.

Inverted indexes

This is where single-table design gets powerful.

Imagine you have users and orders. You want to:

Get all orders for a user (query by user_id)
Find which user made a specific order (query by order_id)

With a regular table, you can only query by the hash key (user_id). To query by order_id, you need a Global Secondary Index (GSI) that "inverts" the keys - the sort key becomes the hash key.

inverted_index.py

"""Example: Inverted index for bidirectional queries."""

import asyncio

from pydynox import Model, ModelConfig
from pydynox.attributes import StringAttribute
from pydynox.indexes import GlobalSecondaryIndex


class UserOrder(Model):
    model_config = ModelConfig(table="app")

    # Main table: pk=USER#123, sk=ORDER#456
    pk = StringAttribute(partition_key=True, template="USER#{user_id}")
    sk = StringAttribute(sort_key=True, template="ORDER#{order_id}")
    user_id = StringAttribute()
    order_id = StringAttribute()
    status = StringAttribute()

    # Inverted index: sk becomes hash key, pk becomes sort key
    by_order = GlobalSecondaryIndex(
        index_name="inverted",
        partition_key="sk",
        sort_key="pk",
    )


async def main():
    # Create orders
    order1 = UserOrder(user_id="alice", order_id="001", status="shipped")
    order2 = UserOrder(user_id="alice", order_id="002", status="pending")
    order3 = UserOrder(user_id="bob", order_id="003", status="shipped")
    await order1.save()
    await order2.save()
    await order3.save()

    # Query by user (main table)
    print("Alice's orders:")
    async for order in UserOrder.query(user_id="alice"):
        print(f"  {order.order_id}: {order.status}")

    # Query by order (inverted index)
    print("\nWho made order 003?")
    async for result in UserOrder.by_order.query(order_id="003"):
        print(f"  User: {result.user_id}")

    # Cleanup
    await order1.delete()
    await order2.delete()
    await order3.delete()


if __name__ == "__main__":
    asyncio.run(main())

The GSI query also uses template placeholders. You pass order_id="003" and pydynox builds ORDER#003 for the GSI hash key.

Follower/following pattern

A classic social media pattern. You want to answer two questions:

Who does Alice follow?
Who follows Bob?

follower_following.py

"""Example: Follower/following social pattern."""

import asyncio

from pydynox import Model, ModelConfig
from pydynox.attributes import StringAttribute
from pydynox.indexes import GlobalSecondaryIndex


class Follow(Model):
    model_config = ModelConfig(table="social")

    # Main: pk=FOLLOWER#alice, sk=FOLLOWING#bob
    pk = StringAttribute(partition_key=True, template="FOLLOWER#{follower}")
    sk = StringAttribute(sort_key=True, template="FOLLOWING#{following}")
    follower = StringAttribute()
    following = StringAttribute()

    # Inverted: pk=FOLLOWING#bob, sk=FOLLOWER#alice
    followers_index = GlobalSecondaryIndex(
        index_name="followers",
        partition_key="sk",
        sort_key="pk",
    )


async def main():
    # Alice follows Bob and Charlie
    f1 = Follow(follower="alice", following="bob")
    f2 = Follow(follower="alice", following="charlie")
    # Dave follows Bob
    f3 = Follow(follower="dave", following="bob")
    await f1.save()
    await f2.save()
    await f3.save()

    # Who does Alice follow? (main table)
    print("Alice follows:")
    async for f in Follow.query(follower="alice"):
        print(f"  {f.following}")

    # Who follows Bob? (inverted index)
    print("\nBob's followers:")
    async for f in Follow.followers_index.query(following="bob"):
        print(f"  {f.follower}")

    # Cleanup
    await f1.delete()
    await f2.delete()
    await f3.delete()


if __name__ == "__main__":
    asyncio.run(main())

One table, one GSI, two access patterns.

Mixed entity types

Multiple models can share the same table. This is the core of single-table design:

mixed_entities.py

"""Example: Multiple entity types in same table."""

import asyncio

from pydynox import Model, ModelConfig
from pydynox.attributes import StringAttribute


class User(Model):
    model_config = ModelConfig(table="app")
    pk = StringAttribute(partition_key=True, template="USER#{user_id}")
    sk = StringAttribute(sort_key=True, template="PROFILE")
    user_id = StringAttribute()
    name = StringAttribute()


class Order(Model):
    model_config = ModelConfig(table="app")
    pk = StringAttribute(partition_key=True, template="USER#{user_id}")
    sk = StringAttribute(sort_key=True, template="ORDER#{order_id}")
    user_id = StringAttribute()
    order_id = StringAttribute()
    total = StringAttribute()


async def main():
    # Create user and orders - all in the same table
    user = User(user_id="alice", name="Alice")
    order1 = Order(user_id="alice", order_id="001", total="100")
    order2 = Order(user_id="alice", order_id="002", total="200")
    await user.save()
    await order1.save()
    await order2.save()

    # Get user profile
    found = await User.get(pk="USER#alice", sk="PROFILE")
    if found:
        print(f"User: {found.name}")

    # Get orders - filter by sk prefix to exclude the profile
    print("Orders:")
    async for order in Order.query(
        user_id="alice",
        sort_key_condition=Order.sk.begins_with("ORDER#"),
    ):
        print(f"  {order.order_id}: ${order.total}")

    # Cleanup
    await user.delete()
    await order1.delete()
    await order2.delete()


if __name__ == "__main__":
    asyncio.run(main())

The begins_with("ORDER#") filter is important. Without it, the query would also return the user's profile (which has the same pk but sk="PROFILE").

Collection - multi-entity query

When you have multiple entity types under the same partition key, you often want to fetch them all in one query. Collection does this for you.

Instead of:

user = await User.get(pk="USER#123", sk="PROFILE")
orders = [o async for o in Order.query(pk="USER#123")]
# = 2 DynamoDB calls

Use Collection:

from pydynox import Collection

collection = Collection([User, Order, Address])
result = await collection.query(pk="USER#123")
# = 1 DynamoDB call

result.users      # [User(...)]
result.orders     # [Order(...), Order(...)]
result.addresses  # [Address(...)]

Requirements

All models in a Collection must:

Share the same table
Have a discriminator field (_type = StringAttribute(discriminator=True))

The discriminator tells pydynox which model class to use when deserializing each item.

Basic usage

basic.py

"""Basic Collection usage - query multiple entity types in one call."""

import asyncio

from pydynox import Collection, Model, ModelConfig
from pydynox.attributes import StringAttribute


class User(Model):
    model_config = ModelConfig(table="app")
    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    _type = StringAttribute(discriminator=True)
    name = StringAttribute()


class Order(Model):
    model_config = ModelConfig(table="app")
    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    _type = StringAttribute(discriminator=True)
    total = StringAttribute()


async def main():
    # Create collection
    collection = Collection([User, Order])

    # Query all entities for a user
    result = await collection.query(pk="USER#123")

    # Access results by type
    for user in result.users:
        print(f"User: {user.name}")

    for order in result.orders:
        print(f"Order total: {order.total}")

    # Or use get() for explicit type
    _users: list[User] = result.get(User)
    _orders: list[Order] = result.get(Order)


if __name__ == "__main__":
    asyncio.run(main())

Filter by sort key

Use sk_begins_with to filter results:

sk_filter.py

"""Collection with sort key filter."""

import asyncio

from pydynox import Collection, Model, ModelConfig
from pydynox.attributes import StringAttribute


class User(Model):
    model_config = ModelConfig(table="app")
    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    _type = StringAttribute(discriminator=True)
    name = StringAttribute()


class Order(Model):
    model_config = ModelConfig(table="app")
    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    _type = StringAttribute(discriminator=True)
    total = StringAttribute()


async def main():
    collection = Collection([User, Order])

    # Get only orders (filter by sk prefix)
    result = await collection.query(pk="USER#123", sk_begins_with="ORDER#")

    # Only orders returned
    print(f"Orders: {len(result.orders)}")
    print(f"Users: {len(result.users)}")  # 0


if __name__ == "__main__":
    asyncio.run(main())

With inheritance

Collection works well with model inheritance:

with_inheritance.py

"""Collection with model inheritance."""

import asyncio

from pydynox import Collection, Model, ModelConfig
from pydynox.attributes import StringAttribute


# Base class with shared attributes
class BaseEntity(Model):
    model_config = ModelConfig(table="app")
    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    _type = StringAttribute(discriminator=True)


class User(BaseEntity):
    name = StringAttribute()
    email = StringAttribute()


class Order(BaseEntity):
    total = StringAttribute()
    status = StringAttribute()


class Address(BaseEntity):
    street = StringAttribute()
    city = StringAttribute()


async def main():
    # All models share the same table and discriminator
    collection = Collection([User, Order, Address])

    result = await collection.query(pk="USER#123")

    # Results are typed correctly
    user: User = result.users[0]
    print(f"User: {user.name} ({user.email})")

    for order in result.orders:
        print(f"Order: ${order.total} - {order.status}")

    for addr in result.addresss:
        print(f"Address: {addr.street}, {addr.city}")


if __name__ == "__main__":
    asyncio.run(main())

Typed results

Results are fully typed. IDE autocomplete works:

result = await collection.query(pk="USER#123")

# IDE knows these are User and Order instances
user: User = result.users[0]
order: Order = result.orders[0]

# Autocomplete works
print(user.name)    # User.name
print(order.total)  # Order.total

Sync version

Use sync_query for synchronous code:

result = collection.sync_query(pk="USER#123")

Best practices

Practice	Why
Use clear prefixes	`USER#`, `ORDER#`, `PROFILE` make debugging easier
Match placeholder names to attributes	`template="USER#{user_id}"` with `user_id = StringAttribute()`
Filter by sk prefix	When querying mixed entities with same pk
Create GSI at table creation	Adding GSIs later requires a migration
Use discriminator for Collection	Required for multi-entity queries

Testing your code

Use MemoryBackend as a pytest fixture to test template keys without DynamoDB:

testing_template.py

"""Example: Testing template keys with MemoryBackend."""

import pytest
from pydynox import Model, ModelConfig
from pydynox.attributes import StringAttribute
from pydynox.indexes import GlobalSecondaryIndex
from pydynox.testing import MemoryBackend


class UserOrder(Model):
    model_config = ModelConfig(table="app")

    pk = StringAttribute(partition_key=True, template="USER#{user_id}")
    sk = StringAttribute(sort_key=True, template="ORDER#{order_id}")
    user_id = StringAttribute()
    order_id = StringAttribute()
    status = StringAttribute()

    by_order = GlobalSecondaryIndex(
        index_name="inverted",
        partition_key="sk",
        sort_key="pk",
    )


@pytest.fixture
def memory_backend():
    with MemoryBackend():
        yield


@pytest.mark.asyncio
async def test_template_key_building(memory_backend):
    # GIVEN an order with template keys
    order = UserOrder(user_id="alice", order_id="001", status="pending")

    # THEN keys are built from template
    assert order.pk == "USER#alice"
    assert order.sk == "ORDER#001"


@pytest.mark.asyncio
async def test_query_with_placeholder(memory_backend):
    # GIVEN saved orders
    await UserOrder(user_id="alice", order_id="001", status="shipped").save()
    await UserOrder(user_id="alice", order_id="002", status="pending").save()

    # WHEN querying by placeholder
    orders = [o async for o in UserOrder.query(user_id="alice")]

    # THEN returns matching orders
    assert len(orders) == 2
    assert orders[0].user_id == "alice"


@pytest.mark.asyncio
async def test_gsi_query_with_placeholder(memory_backend):
    # GIVEN saved orders
    await UserOrder(user_id="alice", order_id="001", status="shipped").save()
    await UserOrder(user_id="bob", order_id="002", status="pending").save()

    # WHEN querying GSI by placeholder
    results = [r async for r in UserOrder.by_order.query(order_id="001")]

    # THEN returns the order owner
    assert len(results) == 1
    assert results[0].user_id == "alice"

Table creation

Create the table with the inverted GSI:

from pydynox import DynamoDBClient

client = DynamoDBClient()

await client.create_table(
    "app",
    partition_key=("pk", "S"),
    sort_key=("sk", "S"),
    global_secondary_indexes=[
        {
            "index_name": "inverted",
            "partition_key": ("sk", "S"),
            "sort_key": ("pk", "S"),
            "projection": "ALL",
        },
    ],
)

Next steps

Indexes - Learn more about GSI and LSI
Query - Query patterns and pagination
Conditions - Conditional writes and filters
Testing - Testing with MemoryBackend