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.