Skip to content

DynamoDB

Single-table utilities (atlas.dynamodb) is the package that provides the foundational building blocks for implementing the Single-Table Design pattern on Amazon DynamoDB within Atlas-based applications.

It defines standardized abstractions for entity modeling and data access, enabling multiple domain entities to coexist safely and efficiently within a single DynamoDB table, while preserving strong consistency rules, clear ownership boundaries, and predictable access patterns.

The package is centered around two core concepts:

SingleTableEntity — declarative representation of a domain entity

SingleTableService — controlled access layer for reading and writing entities

Together, they enforce a clean separation between domain modeling and persistence mechanics, while remaining flexible enough to support complex access patterns, GSIs, and polymorphic records.

SingleTableEntity

Bases: BaseModel

Base model for entities stored using the DynamoDB single-table design pattern.

This class represents a generic item persisted in a single DynamoDB table, where multiple logical entity types coexist and are distinguished through composite primary keys (pk and sk).

Key conventions
  • pk (partition key): Identifies the logical entity instance. If not explicitly provided, it is automatically generated in the format: #

  • sk (sort key): Identifies the record type within the partition. By default, the value "META" is used to represent the primary record of the entity.

This class is designed to be extended by concrete models residing inside the app-edge package dynamodels. All subclasses benefit from: - Automatic primary key generation - Strict schema validation (extra fields are forbidden) - Native conversion to and from DynamoDB item dictionaries

The class intentionally contains no persistence logic; all database interactions are delegated to SingleTableService.

from_item(item) classmethod

Creates an entity instance from a DynamoDB item dictionary.

Parameters:

Name Type Description Default
item Dict[str, Any]

A dictionary representing the item in DynamoDB format.

required

Returns: An instance of the entity populated with data from the item.

to_item()

Converts the entity instance into a DynamoDB item dictionary.

Returns:

Type Description
Dict[str, Any]

A dictionary representing the item in DynamoDB format.

SingleTableService

Persistence service for DynamoDB single-table entities.

This service provides a high-level, opinionated abstraction over DynamoDB operations for entities following the single-table design pattern and extending SingleTableEntity.

Design principles
  • The service is stateless aside from a cached table reference
  • All methods are safe to use concurrently
  • Business logic is intentionally excluded; this service only handles persistence concerns
  • By default, the table name is derived from the current application name: {app_name}_domain.
  • A custom table name and/or AWS region may be provided at initialization, allowing this service to operate on tables outside the current CloudFormation stack or even in another region.

create(entity)

Creates a new entity in the DynamoDB table.

Parameters:

Name Type Description Default
entity SingleTableEntity

The entity to create.

required

Raises: ConflictError: If an entity with the same primary key already exists.

delete(pk, sk='META')

Deletes an entity from the DynamoDB table.

Parameters:

Name Type Description Default
pk str

The partition key of the entity.

required
sk str

The sort key of the entity.

'META'

Raises: NotFoundError: If the entity does not exist.

get(entity_cls, pk, sk='META')

Retrieves an entity from the DynamoDB table.

Parameters:

Name Type Description Default
entity_cls type[T]

The class of the entity to retrieve.

required
pk str

The partition key of the entity.

required
sk str

The sort key of the entity.

'META'

Returns: An instance of the requested entity class. Raises: NotFoundError: If the entity does not exist.

query(pk, sk_prefix=None)

Queries items in the DynamoDB table.

Parameters:

Name Type Description Default
pk str

The partition key to query.

required
sk_prefix Optional[str]

Optional prefix for the sort key to filter results.

None

Returns: A list of items matching the query.

query_entities(entity_cls, pk, sk_prefix=None)

Queries entities in the DynamoDB table.

Parameters:

Name Type Description Default
entity_cls type[T]

The SingleTableEntity extending class of the entities to retrieve.

required
pk str

The partition key to query.

required
sk_prefix Optional[str]

Optional prefix for the sort key to filter results.

None

Returns: A list of instances of the requested entity class.

replace(entity)

Replaces an existing entity in the DynamoDB table.

Parameters:

Name Type Description Default
entity SingleTableEntity

The entity to replace.

required

Raises: NotFoundError: If the entity does not exist.

update_fields(pk, sk='META', updates=None)

Updates specific fields of an existing entity in the DynamoDB table.

Parameters:

Name Type Description Default
pk str

The partition key of the entity.

required
sk str

The sort key of the entity.

'META'
updates Optional[dict[str, Any]]

A dictionary of field names and their new values.

None

Raises: NotFoundError: If the entity does not exist.