Skip to content

DynamoDB & Single-Table Models

Single-table models provide Atlas applications with a DynamoDB persistence option for domains or workloads that do not fit the default SQLAlchemy + PostgreSQL model.

They can be used alone or alongside relational persistence. Use DynamoDB when the data shape is naturally access-pattern-oriented, when locality matters, or when a part of the domain benefits from DynamoDB's operational model.

Mental Model

Atlas' DynamoDB layer separates four ideas:

  • Identity: who the entity is. This is the public id used by application code.
  • Storage address: where the entity lives in DynamoDB. This is the internal pk/sk.
  • Entity type: how the entity is physically modeled, either canonical or aggregate.
  • Access pattern: an alternate read route materialized in the same table.

Application code should talk in public ids:

business_id: str
user_id: str
conversation_id: str

Application code should not model domain relationships using DynamoDB keys:

business_pk: str  # Avoid
user_pk: str      # Avoid

pk and sk are implementation details owned by Atlas.

Every SingleTableEntity has an id field. When an application does not provide one, Atlas generates a UUIDv4 string by default:

550e8400-e29b-41d4-a716-446655440000

The examples below use UUID-looking ids to match the default behavior. You may still provide application-specific ids when your domain needs them.

Initializing Dynamodels

Create a src/dynamodels package to hold DynamoDB-backed domain models:

src/dynamodels/__init__.py
from .agent import Agent
from .business import Business
from .conversation import Conversation
from .message import Message
from .user import User

__all__ = [
    "Agent",
    "Business",
    "Conversation",
    "Message",
    "User",
]

Use one file per model, named in snake case, with a PascalCase class.

Entity Types

Atlas exposes two concrete entity bases:

  • CanonicalEntity
  • AggregateEntity

Both inherit from SingleTableEntity, the common base for Atlas single-table domain entities. Application models should normally inherit from one of the concrete classes.

CanonicalEntity

Use CanonicalEntity by default.

A canonical entity has independent identity and is stored at:

pk = EntityName#id
sk = META

Example:

src/dynamodels/user.py
from atlas.dynamodb import CanonicalEntity, index, unique

class User(CanonicalEntity):
    email: str = unique()
    name: str
    category: str | None = index(default=None)

With id="550e8400-e29b-41d4-a716-446655440000", Atlas stores:

pk = User#550e8400-e29b-41d4-a716-446655440000
sk = META

Good canonical entity examples:

  • User
  • Business
  • Agent
  • Contact
  • Template
  • Integration
  • Conversation

Use canonical entities for things that have their own lifecycle and may participate in multiple relationships or read patterns.

AggregateEntity

Use AggregateEntity only when the entity is the many side of a stable one-to-many relationship and the dominant read path is listing children by parent.

An aggregate entity is stored inside the parent partition:

pk = ParentEntity#parent_id
sk = EntityName#ordered_value#id

Example:

src/dynamodels/message.py
from atlas.dynamodb import AggregateConfig, AggregateEntity, index

from dynamodels import Conversation

class Message(AggregateEntity):
    conversation_id: str
    created_at: str
    sender_user_id: str = index()
    text: str

    aggregate = AggregateConfig(
        parent=Conversation,
        parent_id="conversation_id",
        sort=["created_at"],
    )

With conversation_id="11111111-1111-4111-8111-111111111111", created_at="2026-05-27T22:10:00Z", and id="22222222-2222-4222-8222-222222222222", Atlas stores:

pk = Conversation#11111111-1111-4111-8111-111111111111
sk = Message#2026-05-27T22%3A10%3A00Z#22222222-2222-4222-8222-222222222222

Use aggregate entities when all of these are true:

  • the entity is the many side of a one-to-many relationship;
  • the main query lists many children by one parent;
  • the list has natural ordering and pagination;
  • the collection may become high volume;
  • the parent is stable;
  • parent-partition locality is useful.

Good aggregate examples:

  • Message
  • AuditLog
  • WebhookEvent
  • ConversationEvent
  • Notification

If the child has many independent access paths, a separate lifecycle, or frequently changes parent, prefer CanonicalEntity + index().

AggregateConfig

AggregateConfig tells Atlas how to compute an aggregate entity's storage address.

aggregate = AggregateConfig(
    parent=Conversation,
    parent_id="conversation_id",
    sort=["created_at"],
)

parent is the parent entity class or entity type name. It defines the partition prefix.

parent_id is the field on the aggregate that stores the parent's public id.

sort is the ordered list of fields that appear in the aggregate sort key before the aggregate's own id.

Atlas encodes key parts before composing keys, so values containing separators such as # do not collide with the key format. Aggregate sort fields must be strings or enums inheriting from str, because aggregate sort keys use lexicographic string ordering. Use already normalized sortable values, such as ISO 8601 timestamps.

Ordering

DynamoDB sort keys are compared lexicographically. Choose values that sort correctly as strings.

For dates, use fixed-width UTC timestamps:

2026-05-27T22:10:00.000Z

For numbers, use fixed-width strings if numeric order matters:

0000000042
0000000137

For strings, normalize case and locale-sensitive content in your application when needed.

If you need descending order, prefer querying with scan_forward=False rather than reversing the stored timestamp format.

Access Patterns

Atlas keeps access patterns materialized in the same DynamoDB table. It does not require adding a real GSI for every application query.

The supported declarations are:

  • unique()
  • index()
  • CompositeIndex

These work for both canonical and aggregate entities.

Atlas prefixes each materialized access-pattern value with its scalar type before encoding it. This prevents collisions between values with the same text representation, such as None and "None" or 1 and "1". Enum values are normalized to their underlying scalar value, so callers may query using either the enum instance or its value. Equivalent finite decimals are normalized consistently.

Atlas rejects non-finite decimal values and validates materialized keys against DynamoDB's 2048-byte partition-key and 1024-byte sort-key limits before sending requests to AWS.

unique()

Use unique() when one value may belong to only one entity and should support lookup.

from atlas.dynamodb import CanonicalEntity, unique

class User(CanonicalEntity):
    email: str = unique()
    external_id: str | None = unique(default=None)

Atlas creates an item similar to:

pk = UNIQUE#User#email#str:a%40example.com
sk = TARGET
target_pk = User#550e8400-e29b-41d4-a716-446655440000
target_sk = META

Creating this item with attribute_not_exists makes uniqueness safe under concurrent writes.

When a unique field is None, Atlas does not create an access-pattern item. Multiple entities may therefore have None for the same unique field, matching the usual nullable unique-constraint semantics. unique() intentionally does not provide allow_none=True, because materializing None would allow only one entity to hold it and is rarely a useful domain rule.

index()

Use index() when many entities can share the same value.

from atlas.dynamodb import CanonicalEntity, index

class Agent(CanonicalEntity):
    business_id: str = index()
    user_id: str = index()
    role: str

This supports queries such as:

agents = single_table_service.query_by_index(Agent, "business_id", business_id)

Atlas creates items similar to:

pk = INDEX#Agent.business_id#str:11111111-1111-4111-8111-111111111111
sk = Agent#33333333-3333-4333-8333-333333333333
target_pk = Agent#33333333-3333-4333-8333-333333333333
target_sk = META

CompositeIndex

Use CompositeIndex when you need to query a collection by partition fields and optionally filter/order within that collection by sort fields.

from typing import ClassVar

from atlas.dynamodb import CanonicalEntity, CompositeIndex, SingleTableConfig

class Order(CanonicalEntity):
    customer_id: str
    status: str
    created_at: str

    access_patterns: ClassVar[SingleTableConfig] = SingleTableConfig(
        indexes=[
            CompositeIndex(
                partition=["customer_id"],
                sort=["status", "created_at"],
            )
        ]
    )

Then query:

orders = single_table_service.query_by_index(
    Order,
    "Order.customer_id_status_created_at",
    customer_id,
)

paid_orders = single_table_service.query_by_index(
    Order,
    "Order.customer_id_status_created_at",
    customer_id,
    sort_begins_with={"status": "PAID"},
)

Sort filter dictionaries must follow the declared order. If sort=["status", "created_at"], you may filter by status or by status + created_at, but not by created_at alone.

Composite partition queries must provide exactly the declared partition fields. Composite sort predicates may provide only a non-empty prefix of the declared sort fields. Atlas rejects missing, extra, out-of-order, and excessive values instead of querying a different key silently.

Sort predicates are supported only for CompositeIndex. A CollectionIndex uses its sort key as an internal target address and does not expose it as an application-level ordering contract.

Composite sort fields must be strings or enums inheriting from str, because materialized composite sort keys use DynamoDB's lexicographic string ordering. Use normalized sortable strings such as ISO 8601 timestamps. Integers, booleans, decimals, and integer enums remain supported in unique constraints, collection indexes, and composite partition fields.

When allow_none=False, Atlas does not materialize a composite index if any declared partition or sort field is None. Set allow_none=True only when None should be an explicit part of the materialized key.

With allow_none=True, Atlas encodes None deterministically as null:. A None partition value groups entities into the same partition, while a None sort value participates in lexicographic ordering as that encoded value. This is predictable, but often not the desired query model.

For explicit naming, pass name:

CompositeIndex(
    name="Order.by_customer_status",
    partition=["customer_id"],
    sort=["status", "created_at"],
)

Embedded Models

Access patterns declared explicitly through SingleTableConfig may reference fields in embedded models using dotted paths. The access pattern still belongs to and resolves to the containing entity:

from typing import ClassVar

from pydantic import BaseModel

from atlas.dynamodb import (
    CanonicalEntity,
    CompositeIndex,
    SingleTableConfig,
    UniqueConstraint,
)

class WhatsAppBusinessAccount(BaseModel):
    business_account_id: str
    phone_number_id: str

class Workspace(CanonicalEntity):
    whatsapp: WhatsAppBusinessAccount | None = None

    access_patterns: ClassVar[SingleTableConfig] = SingleTableConfig(
        unique=[
            UniqueConstraint(field="whatsapp.phone_number_id"),
        ],
        indexes=[
            CompositeIndex(
                name="Workspace.by_whatsapp_business_account",
                partition=["whatsapp.business_account_id"],
            ),
        ],
    )

Atlas does not discover unique() or index() declarations inside embedded models. Declare embedded-field access patterns explicitly on the containing entity. If an intermediate embedded value is None, unique constraints are omitted, while indexes apply their configured allow_none behavior. Dotted paths traverse object attributes through embedded Pydantic models; they do not traverse scalar values, list elements, or dictionary keys.

Atlas validates declared access patterns against the entity's Pydantic schema before using them. Paths must exist, intermediate fields must be embedded Pydantic models, and terminal fields must declare one supported scalar type, optionally combined with None. Supported scalar types are strings, integers, booleans, decimals, and enums that inherit from str or int. Use Decimal, not float, for DynamoDB numeric fields. Store dates, datetimes, and UUIDs as normalized strings.

Access-pattern keys preserve scalar types to avoid collisions. Query values must therefore use the same scalar type declared by the entity field: query a Decimal field with Decimal, an integer field with int, and so on. String and integer enum members normalize to their underlying scalar values.

Access-pattern values are materialized in auxiliary item keys and attributes; they are encoded, not hashed or encrypted. Never declare tokens, secrets, credentials, or other sensitive values as unique constraints or indexes.

Atlas also rejects empty composite partitions, repeated fields, and ambiguous field/name identifiers across unique constraints or indexes. Runtime values are checked again before key materialization, preventing mutable objects and collections from becoming unstable key strings.

Embedded-field access patterns behave like access patterns over direct fields:

  • create() creates their materialized access-pattern items;
  • replace() updates those items when an embedded value changes;
  • delete() removes them;
  • get_by_unique() and query_by_index() resolve to the containing entity.

For the Workspace model above:

workspace = single_table_service.get_by_unique(
    Workspace,
    "whatsapp.phone_number_id",
    phone_number_id,
)

workspaces = single_table_service.query_by_index(
    Workspace,
    "Workspace.by_whatsapp_business_account",
    business_account_id,
)

Both access patterns return Workspace instances, not independent WhatsAppBusinessAccount instances. An embedded model has no storage address or lifecycle of its own.

Invalid dotted paths raise DatabaseError before Atlas uses the access pattern. Adding an access pattern does not backfill existing entities, and removing or renaming one does not clean up its existing materialized items. Plan a data migration whenever these declarations change after entities have already been persisted.

Relationships

Relationships are modeled with public ids and explicit access patterns.

One-to-One

Use unique() when one side must be owned by only one entity.

class BusinessProfile(CanonicalEntity):
    business_id: str = unique()
    display_name: str

Read:

profile = single_table_service.get_by_unique(BusinessProfile, "business_id", business_id)

Many-to-One

The many side stores the one side's public id and marks it with index().

class Agent(CanonicalEntity):
    business_id: str = index()
    user_id: str = index()

Read:

business_agents = single_table_service.query_by_index(Agent, "business_id", business_id)

One-to-Many

Use CanonicalEntity + index() when children are independent:

class Contact(CanonicalEntity):
    business_id: str = index()
    name: str

Use AggregateEntity when children are mostly listed by parent:

class Message(AggregateEntity):
    conversation_id: str
    created_at: str
    text: str

    aggregate = AggregateConfig(
        parent=Conversation,
        parent_id="conversation_id",
        sort=["created_at"],
    )

Many-to-Many

Create an entity for the relationship.

class UserGroupMembership(CanonicalEntity):
    user_id: str = index()
    group_id: str = index()
    role: str

Use a relationship entity when the relation has metadata, lifecycle, permissions, timestamps, or needs to be queried from both sides.

Persistence API

Use single_table_service from Atlas' DynamoDB DI container or instantiate SingleTableService directly for advanced cases.

Create:

single_table_service.create(user)

Read by public id:

user = single_table_service.get_by_id(User, "550e8400-e29b-41d4-a716-446655440000")
message = single_table_service.get_by_id(Message, "22222222-2222-4222-8222-222222222222")

For aggregate entities, Atlas maintains an internal unique id lookup item so the caller does not need to provide parent id or sort values.

Atlas intentionally keeps DynamoDB pk/sk out of the normal read API. If infrastructure-level code needs to read by storage address, use get_by_address() explicitly.

Read by unique field:

user = single_table_service.get_by_unique(User, "email", email)

Query by index:

agents = single_table_service.query_by_index(Agent, "business_id", business_id)

Query one page:

page = single_table_service.query_by_index_page(
    Agent,
    "business_id",
    business_id,
    limit=25,
    next_token=request_next_token,
)

return {
    "items": page.items,
    "next_token": page.next_token,
}

Methods that return a plain list, such as query_by_index() and list_by_parent(), automatically consume every DynamoDB page instead of silently stopping at DynamoDB's 1 MB response boundary. Prefer their *_page counterparts for potentially large or unbounded collections so callers control memory use, latency, and consumed capacity.

Public Atlas reads use DynamoDB's eventually consistent reads by default. This matches DynamoDB's normal behavior and avoids paying the higher read-capacity cost of strong consistency for every request.

Use strong_reads() when a complete domain operation requires strong reads:

with single_table_service.strong_reads():
    user = user_service.get_by_email(email)
    workspace = workspace_service.get_by_id(workspace_id)

The policy applies to reads performed through any SingleTableService instance in the current execution context. Atlas implements it with ContextVar, so it is isolated between independently running requests and asynchronous tasks. A child task created inside the block inherits the policy active at its creation. The policy is restored when the block exits, including after exceptions, and nested policies restore the previous value:

with single_table_service.strong_reads():
    user = single_table_service.get_by_id(User, user_id)  # strong

    messages = single_table_service.list_by_parent(
        Message,
        conversation_id,
        consistent_read=False,
    )  # eventual

    workspace = single_table_service.get_by_id(Workspace, workspace_id)  # strong

Avoid starting detached background tasks inside a strong policy block unless they should also use strong reads: a child task keeps the policy context it inherited when it was created, even if it finishes after the parent block.

Pass consistent_read=True or consistent_read=False when one specific call must override the contextual policy:

user = single_table_service.get_by_id(
    User,
    user_id,
    consistent_read=True,
)

When consistent_read is omitted or None, the read inherits the contextual policy; outside a policy block it remains eventual. Explicit overrides are available on point reads, index queries, aggregate listings, and raw reads. Atlas explicitly uses strong reads internally where entity writes need the current persisted state for access-pattern maintenance and optimistic locking.

Strong reads make each individual read observe all writes completed before that read began. A strong_reads() block is not a transactional snapshot: another writer can still change data between two reads in the same block. When a business invariant must remain true until a write commits, enforce it with an optimistic-lock condition, tx.condition_check(), or another operation in the same DynamoDB transaction.

Do not rely on the usually short eventual-consistency window for correctness- critical decisions. Use a strong_reads() block or an explicit consistent_read=True for authentication, authorization, relationship or ownership checks, write preconditions, and workflows that must immediately observe a completed write. Keep policy blocks narrow so unrelated reads do not incur strong-read cost. Eventual reads are appropriate for ordinary lookups and listings that can tolerate briefly stale or missing results.

Index collection queries resolve materialized access-pattern items to their target entities in a second read. A target can disappear because of a concurrent delete or be temporarily invisible to an eventual read. In that case, query_by_index() and query_by_index_page() omit that target instead of failing the complete collection query. They do not remove the access-pattern item automatically because the missing target may be temporary.

An eventual read can also observe an old access-pattern item after the target entity's indexed value changed. Atlas confirms that each resolved entity still materializes one of the access-pattern items returned by the query and omits stale mismatches, preventing an entity that no longer matches the query from being returned. It also suppresses duplicate targets if old and new index items for the same entity are temporarily visible together.

Consequently, an index page can contain fewer resolved entities than its requested limit. Its next_token still points to the correct next access- pattern page. Singular reads such as get_by_unique() remain strict and raise NotFoundError when their access-pattern item or target cannot be read, or when an eventual read resolves a target that no longer has the queried unique value.

List aggregate children:

messages = single_table_service.list_by_parent(Message, conversation_id)

List one aggregate page:

page = single_table_service.list_by_parent_page(
    Message,
    conversation_id,
    limit=50,
    next_token=request_next_token,
    scan_forward=True,
)

Filter aggregate sort values:

messages = single_table_service.list_by_parent(
    Message,
    conversation_id,
    sort_between=(
        {"created_at": start_timestamp},
        {"created_at": end_timestamp},
    ),
)

Updates

Use replace() to update an entity while keeping access patterns consistent:

user = single_table_service.get_by_id(User, user_id)
updated_user = user.model_copy(update={"email": new_email})
single_table_service.replace(updated_user)

replace() compares the current entity with the next entity and updates changed access-pattern items transactionally.

Atlas automatically applies optimistic locking to high-level entity writes. Each persisted entity contains an internal _version attribute. Reads preserve that version privately on the Pydantic entity, while model_dump(), DTOs, and ModelMapper do not expose or copy it as a domain field.

create() starts at version 1. replace() writes the next version only when the persisted version still matches the version loaded with the provided entity. Concurrent or stale replacements fail with ConflictError instead of silently overwriting another write. Atlas does not retry automatically: reload the entity, reapply the intended change, and retry only when that operation is safe to repeat.

Because replace(entity) requires the loaded version, use the persisted entity or a model_copy() of it. Do not construct a new entity with an existing id and pass it to replace(); it has no persisted version to protect.

The following fields are immutable under replace():

  • id;
  • for aggregate entities, the configured parent_id;
  • for aggregate entities, every configured sort field.

Changing those fields changes the DynamoDB storage address. Model that as delete-and-create, not as a normal update.

Deletes

Use delete() to remove the entity and every access-pattern item Atlas can derive from it:

message = single_table_service.get_by_id(Message, message_id)
single_table_service.delete(message)

delete() reloads the persisted entity before building the transaction so access-pattern items are removed from current stored values. It deletes them only when the persisted entity still matches the provided entity's loaded version; stale deletes fail with ConflictError. tx.delete(entity) does not perform reads, so callers using an explicit transaction must provide the current persisted entity.

DynamoDB domain ids are not foreign keys. Atlas does not infer relationships, prevent deletion of referenced entities, or cascade deletes to other entity types. Domain services must explicitly define the desired policy and use a transaction when related entities must be deleted or changed atomically.

Transactions

Use a transaction when multiple writes must succeed or fail together:

with single_table_service.transaction(client_request_token=request_id):
    user_service.create(user_input)
    business_service.create(business_input)

Inside the block, high-level and raw write methods called through any compatible SingleTableService instance automatically join the active transaction. This allows a transaction opened by one domain service to propagate through nested service calls without adding a tx parameter to every method. Compatible instances must point to the same table and region.

Nested transaction() blocks use REQUIRED propagation: they join the existing transaction, only the root block commits, and a failure in any nested block marks the complete transaction as failed even if the exception is caught later. Atlas does not provide independent nested transactions, savepoints, or REQUIRES_NEW.

The returned transaction object remains available for explicit condition checks and advanced operations:

with single_table_service.transaction() as tx:
    tx.condition_exists(user.storage_address().pk)
    single_table_service.create(workspace)

Transaction entity methods maintain access patterns just like the service methods and apply optimistic locking. tx.replace(current_entity, next_entity) uses the version carried by current_entity; after a successful commit, Atlas updates next_entity to its new version. Entity versions are updated in memory only after the entire transaction succeeds.

Available transaction operations include:

  • tx.create(entity)
  • tx.replace(current_entity, next_entity)
  • tx.delete(entity)
  • tx.create_raw(entity)
  • tx.replace_raw(entity)
  • tx.update_fields_raw(pk, sk="META", updates={...})
  • tx.delete_raw(pk, sk="META")
  • tx.condition_exists(pk, sk="META")
  • tx.condition_not_exists(pk, sk="META")
  • tx.condition_check(...)

Transactions contain writes and condition checks, not reads. Reads inside the block use the configured eventual or strong consistency policy, but only observe committed data. Atlas rejects a point read of an item that already has a pending write in the active transaction because DynamoDB cannot provide read-your-own-writes for it. Collection queries continue to return committed data and do not include pending writes. Keep the in-memory entity while inside the block, or read after the root block commits:

with single_table_service.transaction():
    single_table_service.create(message)

with single_table_service.strong_reads():
    persisted_message = single_table_service.get_by_id(Message, message.id)

Do not start detached asynchronous work inside a transaction block. Tasks created in the block inherit its contextual transaction; all participating work must finish before the root block exits. Do not manually share the returned transaction object between independent execution contexts; open the root block once and let nested calls join it contextually.

DynamoDB transactions are limited to 100 operations. Remember that each declared access pattern adds write operations. Normal DynamoDB limits still apply, including the 400 KB item-size limit and the 4 MB aggregate transaction-size limit. Atlas validates operation count and access-pattern key sizes before sending a transaction; DynamoDB remains the authority for serialized item and transaction size.

Raw Operations

Raw operations address DynamoDB items directly and do not maintain access patterns:

raw_item = single_table_service.get_raw(pk, sk)
entity = single_table_service.get_by_address(User, storage_address)
single_table_service.create_raw(entity)
single_table_service.update_fields_raw(pk, sk, updates={"status": "DONE"})
single_table_service.delete_raw(pk, sk)

Prefer high-level entity methods for application data. Treat get_raw(), get_by_address(), and the *_raw write methods as infrastructure APIs. Do not use raw writes for entities with fields declared through unique(), index(), CompositeIndex, or aggregate storage fields unless you also maintain every related item yourself.

Raw replacements, updates, and deletes bypass optimistic locking conditions and do not increment _version. Raw entity creation still initializes _version, and raw entity replacement requires one so it cannot accidentally corrupt the item shape. Mixing raw and high-level writes for the same entity invalidates stale-write protection.

Optimistic locking requires every existing entity item to contain a positive integer _version. Recreate the table or migrate existing items when upgrading from a version of Atlas that did not persist this metadata.

Value Objects

Objects that do not need independent persistence should remain embedded Pydantic models:

from pydantic import BaseModel

class Address(BaseModel):
    street: str
    city: str

class Business(CanonicalEntity):
    document: str = unique()
    address: Address

Value objects are stored inline inside the parent entity item.