py#dynamodb
This generator creates a new Python project backed by Amazon DynamoDB, using PynamoDB for entity modelling. It generates the application code and infrastructure needed to provision and manage a DynamoDB table using AWS CDK or Terraform, with single-table design support and built-in local development via DynamoDB Local.
Generate a DynamoDB Project
Section titled “Generate a DynamoDB Project”- Install the Nx Console VSCode Plugin if you haven't already
- Open the Nx Console in VSCode
- Click
Generate (UI)in the "Common Nx Commands" section - Search for
@aws/nx-plugin - py#dynamodb - Fill in the required parameters
- Click
Generate
pnpm nx g @aws/nx-plugin:py#dynamodbyarn nx g @aws/nx-plugin:py#dynamodbnpx nx g @aws/nx-plugin:py#dynamodbbunx nx g @aws/nx-plugin:py#dynamodbYou can also perform a dry-run to see what files would be changed
pnpm nx g @aws/nx-plugin:py#dynamodb --dry-runyarn nx g @aws/nx-plugin:py#dynamodb --dry-runnpx nx g @aws/nx-plugin:py#dynamodb --dry-runbunx nx g @aws/nx-plugin:py#dynamodb --dry-runOptions
Section titled “Options”| Parameter | Type | Default | Description |
|---|---|---|---|
| name Required | string | - | Name of the DynamoDB project to generate |
| directory | string | packages | The directory to store the project in. |
| subDirectory | string | - | The sub directory the project is placed in. By default this is the project name. |
| framework | pynamodb | pynamodb | The framework to use for DynamoDB entities. |
| tableName | string | - | The DynamoDB table name. Auto-generated if not specified. |
| infra | dynamodb | none | dynamodb | Infrastructure to provision for the DynamoDB table. |
| iac | inherit | cdk | terraform | inherit | The preferred IaC provider. By default this is inherited from your initial selection. |
| preferInstallDependencies | boolean | true | Whether to prefer installing dependencies after the generator runs. Set to false to defer installing when batching multiple generators (an install still runs if needed so subsequent generators can compute the Nx project graph); install once at the end. |
Generator Output
Section titled “Generator Output”The generator creates the following project structure in the <directory>/<name> directory:
Directory<name>
- __init__.py Package exports
- client.py DynamoDB client and table name resolution
Directoryentities
- base.py Base PynamoDB model with GSI declarations
- example.py Example entity definition
- __init__.py Entity exports
- config.json Table configuration including GSI definitions and local development settings
- project.json Project configuration and build targets
The local development scripts are shared across all DynamoDB projects (both TypeScript and Python) and generated once into:
Directorypackages/common/scripts/src/dynamodb
- create-local-table.ts Creates the DynamoDB table in the local DynamoDB Local instance
- pull-image.ts Pulls the DynamoDB Local image
- start-container.ts Starts the DynamoDB Local container
Infrastructure
Section titled “Infrastructure”Since this generator vends infrastructure as code based on your chosen iacProvider, it will create a project in packages/common which includes the relevant CDK constructs or Terraform modules.
The common infrastructure as code project is structured as follows:
Directorypackages/common/constructs
Directorysrc
Directoryapp/ Constructs for infrastructure specific to a project/generator
- …
Directorycore/ Generic constructs which are reused by constructs in
app- …
- index.ts Entry point exporting constructs from
app
- project.json Project build targets and configuration
Directorypackages/common/terraform
Directorysrc
Directoryapp/ Terraform modules for infrastructure specific to a project/generator
- …
Directorycore/ Generic modules which are reused by modules in
app- …
- project.json Project build targets and configuration
Directorypackages/common/constructs/src
Directoryapp
Directorydynamodb
- <name>.ts Infrastructure specific to your table
Directorycore
- dynamodb.ts Generic DynamoDB table construct
Directorypackages/common/terraform/src
Directoryapp
Directorydynamodb
Directory<name>
- <name>.tf Module specific to your table
Directorycore
Directorydynamodb
- dynamodb.tf Generic DynamoDB module
Local Development
Section titled “Local Development”Starting Local DynamoDB
Section titled “Starting Local DynamoDB”The generator configures a dev target that starts a DynamoDB Local instance and creates the table. Use the project’s dev target:
pnpm nx dev <project-name>yarn nx dev <project-name>npx nx dev <project-name>bunx nx dev <project-name>This automatically:
- Pulls the DynamoDB Local image (
pull-imagetarget) - Starts a container
- Creates a local table with the indexes defined in
config.json
Data Modelling
Section titled “Data Modelling”The generated project uses PynamoDB for entity modelling. All entities must inherit from the generated BaseModel — it resolves the correct DynamoDB table name at runtime, reading from AWS AppConfig when deployed or from config.json when running locally via DynamoDB Local. Without this, PynamoDB will not know which table to use. BaseModel also uses PynamoDB’s polymorphism support to store multiple entity types in a single table, following DynamoDB’s single-table design.
Add or update entity files under <name>/entities/, using the generated example entity as a starting point:
from collections.abc import Iteratorfrom datetime import UTC, datetimefrom pynamodb.attributes import UnicodeAttributefrom .base import BaseModel
class ExampleModel(BaseModel, discriminator='ExampleModel'): """ Key design: pk=EXAMPLE#<id>, sk=EXAMPLE#<id> gsi1pk=CATEGORY#<cat>, gsi1sk=EXAMPLE#<id> <- list items by category gsi2pk=EXAMPLE, gsi2sk=<created_at> <- list all items by date """
name = UnicodeAttribute() category = UnicodeAttribute() created_at = UnicodeAttribute() updated_at = UnicodeAttribute()
@classmethod def make_pk(cls, id: str) -> str: return f'EXAMPLE#{id}'
@classmethod def create(cls, id: str, name: str, category: str) -> 'ExampleModel': now = datetime.now(UTC).isoformat() item = cls( pk=cls.make_pk(id), sk=cls.make_pk(id), gsi1pk=f'CATEGORY#{category}', gsi1sk=cls.make_pk(id), gsi2pk='EXAMPLE', gsi2sk=now, name=name, category=category, created_at=now, updated_at=now, ) item.save() return item
# ── Primary index ───────────────────────────────────────────────────────── @classmethod def get_by_id(cls, id: str) -> 'ExampleModel': return cls.get(cls.make_pk(id), cls.make_pk(id))
# ── gsi1_index: partition=category, sort=id ─────────────────────────────── @classmethod def list_by_category(cls, category: str) -> Iterator['ExampleModel']: return cls.gsi1_index.query(f'CATEGORY#{category}')
# ── gsi2_index: partition=type, sort=created_at ─────────────────────────── @classmethod def list_created_between(cls, start: datetime, end: datetime) -> Iterator['ExampleModel']: return cls.gsi2_index.query( 'EXAMPLE', range_key_condition=ExampleModel.gsi2sk.between( start.isoformat(), end.isoformat(), ), scan_index_forward=False, )For more details, see the PynamoDB tutorial.
Designing Around Access Patterns
Section titled “Designing Around Access Patterns”In DynamoDB, schema design starts with your queries, not your data shape. Before writing any model, list every access pattern your application needs, then design pk, sk, and GSI key values so each pattern is answered by a single table request — no JOINs, no sequential reads.
The generated ExampleModel demonstrates this for three patterns:
- Get by ID — primary index,
pk=EXAMPLE#<id>,sk=EXAMPLE#<id> - List by category —
gsi1,pk=CATEGORY#<category> - List by creation date —
gsi2,pk=EXAMPLE, sort key between ISO timestamps
The type prefix convention (e.g. EXAMPLE#, CATEGORY#) is deliberate: it makes items self-describing when browsing the table, prevents accidental key collisions between entity types that share an index, and allows sort key prefix filtering using begins_with.
Before writing a new entity, define its key patterns upfront in a docstring. The OrderModel in the next section follows this convention:
class OrderModel(BaseModel, discriminator='OrderModel'): """ Key design: pk=ORDER#<order_id>, sk=ORDER#<order_id> gsi1pk=USER#<user_id>, gsi1sk=ORDER#<order_id> <- list orders for a user gsi2pk=ORDER, gsi2sk=<created_at> <- list all orders by date """Storing Multiple Entity Types
Section titled “Storing Multiple Entity Types”PynamoDB’s DiscriminatorAttribute stores a type label (entity_type) in every item. When querying via BaseModel, this label is used to instantiate each result as its correct subclass automatically — so a single query can return a mix of UserModel, OrderModel, and any other entity type registered in the same table.
Below is a complete two-entity example — a UserModel with associated OrderModel records stored in the same table:
from collections.abc import Iteratorfrom datetime import UTC, datetimefrom pynamodb.attributes import UnicodeAttributefrom .base import BaseModel
class UserModel(BaseModel, discriminator='UserModel'): """ Key design: pk=USER#<user_id>, sk=USER#<user_id> gsi2pk=USER, gsi2sk=<created_at> <- list all users by date """
username = UnicodeAttribute() email = UnicodeAttribute() created_at = UnicodeAttribute()
@classmethod def make_pk(cls, user_id: str) -> str: return f'USER#{user_id}'
@classmethod def create(cls, user_id: str, username: str, email: str) -> 'UserModel': now = datetime.now(UTC).isoformat() item = cls( pk=cls.make_pk(user_id), sk=cls.make_pk(user_id), gsi2pk='USER', gsi2sk=now, username=username, email=email, created_at=now, ) item.save() return item
@classmethod def get_by_id(cls, user_id: str) -> 'UserModel': return cls.get(cls.make_pk(user_id), cls.make_pk(user_id))
@classmethod def list_recent(cls, limit: int | None = None) -> Iterator['UserModel']: return cls.gsi2_index.query('USER', limit=limit, scan_index_forward=False)from collections.abc import Iteratorfrom datetime import UTC, datetimefrom pynamodb.attributes import UnicodeAttributefrom .base import BaseModel
class OrderModel(BaseModel, discriminator='OrderModel'): """ Key design: pk=ORDER#<order_id>, sk=ORDER#<order_id> gsi1pk=USER#<user_id>, gsi1sk=ORDER#<order_id> <- list orders by user gsi2pk=ORDER, gsi2sk=<created_at> <- list all orders by date """
user_id = UnicodeAttribute() total = UnicodeAttribute() created_at = UnicodeAttribute()
@classmethod def make_pk(cls, order_id: str) -> str: return f'ORDER#{order_id}'
@classmethod def create(cls, order_id: str, user_id: str, total: str) -> 'OrderModel': now = datetime.now(UTC).isoformat() item = cls( pk=cls.make_pk(order_id), sk=cls.make_pk(order_id), gsi1pk=f'USER#{user_id}', gsi1sk=cls.make_pk(order_id), gsi2pk='ORDER', gsi2sk=now, user_id=user_id, total=total, created_at=now, ) item.save() return item
@classmethod def get_by_id(cls, order_id: str) -> 'OrderModel': return cls.get(cls.make_pk(order_id), cls.make_pk(order_id))
# ── gsi1_index: partition=user, sort=order_id ──────────────────────────── @classmethod def list_by_user(cls, user_id: str) -> Iterator['OrderModel']: return cls.gsi1_index.query(f'USER#{user_id}')
# ── gsi2_index: partition=type, sort=created_at ─────────────────────────── @classmethod def list_recent(cls, limit: int | None = None) -> Iterator['OrderModel']: return cls.gsi2_index.query('ORDER', limit=limit, scan_index_forward=False)Export the new entities from __init__.py:
from .user import UserModelfrom .order import OrderModelfrom .example import ExampleModelGSI Overloading
Section titled “GSI Overloading”BaseModel provides two shared GSIs (gsi1_index, gsi2_index). Both UserModel and OrderModel above write to gsi2 — but with different gsi2pk values (USER vs ORDER). This is GSI overloading: reusing a single physical index to serve multiple independent access patterns without consuming extra GSI capacity.
UserModel—gsi2pk=USER,gsi2sk=<created_at>→ list all users by dateOrderModel—gsi2pk=ORDER,gsi2sk=<created_at>→ list all orders by date
gsi1 can also be overloaded when multiple entity types share the same parent. If you later add a ReviewModel that also belongs to a user, you can assign it gsi1pk=USER#<user_id> with a REVIEW#<id> sort key — no additional GSI needed. Querying gsi1 via BaseModel then returns both orders and reviews for that user in one request, with PynamoDB instantiating each item as its correct subclass:
from .base import BaseModelfrom .order import OrderModelfrom .review import ReviewModel
user_id = 'user-123'items = list(BaseModel.gsi1_index.query(f'USER#{user_id}'))
orders = [i for i in items if isinstance(i, OrderModel)]reviews = [i for i in items if isinstance(i, ReviewModel)]To retrieve only one entity type from an overloaded GSI, use a sort key prefix condition:
orders_only = list(BaseModel.gsi1_index.query( f'USER#{user_id}', range_key_condition=BaseModel.gsi1sk.startswith('ORDER#'),))One-to-Many Relationships
Section titled “One-to-Many Relationships”In a one-to-many relationship the child entity stores a reference to its parent in a GSI partition key, making the relationship traversable in both directions without duplicating data. The UserModel / OrderModel example above is exactly this pattern:
- Get a single order by ID — primary table:
pk=ORDER#<id>,sk=ORDER#<id> - List all orders for a user —
gsi1:pk=USER#<user_id>
An alternative to GSI-based lookups is the item collection pattern: give child items the same pk as their parent and use the sort key to differentiate them. This lets you retrieve the parent and all its children in a single primary-table query, without a GSI:
class OrderModel(BaseModel, discriminator='OrderModel'): """ Key design (item collection): pk=USER#<user_id>, sk=ORDER#<order_id> <- co-located under the parent user """ ...# Retrieve the user and all their orders in one primary-table query# BaseModel dispatches each item to its correct subclass via DiscriminatorAttributeitems = list(BaseModel.query(f'USER#{user_id}'))user = next(i for i in items if isinstance(i, UserModel))orders = [i for i in items if isinstance(i, OrderModel)]The tradeoff: item collections place all children under a single partition key, which is optimal for most workloads but can create a hot partition at extreme write throughput. The GSI approach (used in the examples above) keeps each entity in its own partition and is generally safer to start with.
Many-to-Many Relationships
Section titled “Many-to-Many Relationships”Many-to-many relationships require a junction entity using the adjacency list pattern: a dedicated item that records each link, with its GSI key inverting the direction so the relationship can be traversed both ways.
Consider ArticleModel and TagModel, where an article can have many tags and a tag can apply to many articles:
from collections.abc import Iteratorfrom pynamodb.attributes import UnicodeAttributefrom .base import BaseModel
class ArticleTagModel(BaseModel, discriminator='ArticleTag'): """ Junction entity for the Article ↔ Tag many-to-many relationship.
Key design: pk=ARTICLE#<article_id>, sk=TAG#<tag_name> <- list tags for an article gsi1pk=TAG#<tag_name>, gsi1sk=ARTICLE#<article_id> <- list articles for a tag """
article_id = UnicodeAttribute() tag_name = UnicodeAttribute()
@classmethod def add(cls, article_id: str, tag_name: str) -> 'ArticleTagModel': item = cls( pk=f'ARTICLE#{article_id}', sk=f'TAG#{tag_name}', gsi1pk=f'TAG#{tag_name}', gsi1sk=f'ARTICLE#{article_id}', article_id=article_id, tag_name=tag_name, ) item.save() return item
@classmethod def remove(cls, article_id: str, tag_name: str) -> None: cls.get(f'ARTICLE#{article_id}', f'TAG#{tag_name}').delete()
# ── Primary index: pk=article, sk=tag ───────────────────────────────────── @classmethod def list_tags_for_article(cls, article_id: str) -> Iterator['ArticleTagModel']: return cls.query(f'ARTICLE#{article_id}')
# ── gsi1_index: pk=tag, sk=article ──────────────────────────────────────── @classmethod def list_articles_for_tag(cls, tag_name: str) -> Iterator['ArticleTagModel']: return cls.gsi1_index.query(f'TAG#{tag_name}')Because ArticleTagModel uses pk=ARTICLE#<article_id> — the same partition as the article itself — you can retrieve an article and all its tags in a single primary-table query:
from .base import BaseModelfrom .article import ArticleModelfrom .article_tag import ArticleTagModel
items = list(BaseModel.query('ARTICLE#article-123'))article = next(i for i in items if isinstance(i, ArticleModel))tags = [i.tag_name for i in items if isinstance(i, ArticleTagModel)]For further reading on DynamoDB data modelling, see the DynamoDB data modelling guide and Creating a single-table design with Amazon DynamoDB.
Using the DynamoDB Client
Section titled “Using the DynamoDB Client”The generated client.py exports two key utilities:
is_local()— returnsTruewhenLOCAL_DEV=true, used to switch between local and AWS behaviour.get_table_name()— returns the DynamoDB table name. WhenLOCAL_DEV=true, reads the table name fromlocalDev.tableNameinconfig.json; otherwise fetches the name from AWS AppConfig using theRUNTIME_CONFIG_APP_IDenvironment variable and caches it for subsequent calls.
BaseModel in entities/base.py uses both to configure PynamoDB automatically:
- Connection —
BaseModel.Metasetshostfromconfig.jsonand hardcodesregion,aws_access_key_id, andaws_secret_access_keywhenis_local()isTrue, pointing PynamoDB at the local DynamoDB instance. In AWS, these are left unset so PynamoDB uses the default credential chain. - Table name —
BaseModel._get_connection()callsget_table_name()before each operation, so the correct table is resolved at runtime without any manual configuration.
Stopping Local DynamoDB
Section titled “Stopping Local DynamoDB”Stopping dev (e.g. with Ctrl+C) automatically removes the DynamoDB Local container, but preserves the named volume so your data persists across restarts.
Adding/Removing Global Secondary Indexes
Section titled “Adding/Removing Global Secondary Indexes”GSIs are defined in config.json at the project root under the tableConfig.globalSecondaryIndexes key. Add an entry for each GSI, then reflect the change in BaseModel by adding or removing the corresponding GlobalSecondaryIndex class and attributes in <name>/entities/base.py:
{ ... "tableConfig": { "globalSecondaryIndexes": [ { "indexName": "gsi1pk-gsi1sk-index", "partitionKey": "gsi1pk", "sortKey": "gsi1sk" }, { "indexName": "gsi2pk-gsi2sk-index", "partitionKey": "gsi2pk", "sortKey": "gsi2sk" } ] }}The sortKey field is optional for hash-key-only GSIs.
This config file is the single source of truth read by all consumers:
- Local development —
devreadsconfig.jsonand creates or updates the local table to match the GSI list - CDK — the construct reads
config.jsonat synth time, so GSI changes are reflected on the nextcdk deploy - Terraform — the module reads
config.jsonat plan/apply time
One GSI per Deployment
Section titled “One GSI per Deployment”Connecting to the Table
Section titled “Connecting to the Table”In any Python project, add the DynamoDB package as a workspace dependency and import entity classes directly:
from my_db_package.entities import ExampleModel
item = ExampleModel.get_by_id('123')Deploying your Table
Section titled “Deploying your Table”The DynamoDB generator creates CDK or Terraform infrastructure based on your selected iac.
The CDK construct is created in common/constructs. Example usage:
import { MyTable } from ':my-scope/common-constructs';
export class ApplicationStack extends Stack { constructor(scope: Construct, id: string, props?: StackProps) { super(scope, id, props);
const table = new MyTable(this, 'Table'); }}This provisions a DynamoDB table with:
pk(partition key) andsk(sort key), bothStringtype- Global Secondary Indexes as defined in
config.json - On-demand (
PAY_PER_REQUEST) billing - Customer-managed KMS encryption with automatic key rotation
- Point-in-time recovery enabled
- Deletion protection enabled
- Table name registered in Runtime Config under the
dynamodbnamespace in AWS AppConfig
The Terraform module is created in common/terraform. Example usage:
module "my_table" { source = "../../common/terraform/src/app/dynamodb/my-table"}This provisions a DynamoDB table with:
pk(partition key) andsk(sort key), bothStringtype- Global Secondary Indexes as defined in
config.json - On-demand (
PAY_PER_REQUEST) billing - Customer-managed KMS encryption with automatic key rotation
- Point-in-time recovery enabled
- Deletion protection enabled
- Table name registered in Runtime Config
Granting Access
Section titled “Granting Access”To allow Lambda functions to access the DynamoDB table, grant the necessary permissions in your infrastructure.
Call grantReadWriteData on the table construct. This grants both the DynamoDB and KMS permissions required by the Lambda execution role:
import { MyTable } from ':my-scope/common-constructs';
const table = new MyTable(this, 'Table');
const api = new Api(this, 'Api', { integrations: Api.defaultIntegrations(this).build(),});
Object.entries(api.integrations).forEach(([, integration]) => { table.grantReadWriteData(integration.handler);});Grant the Lambda execution role permission to access the DynamoDB table and its KMS encryption key:
module "my_table" { source = "../../common/terraform/src/app/dynamodb/my-table"}
resource "aws_iam_role_policy" "dynamodb_access" { role = module.my_api.lambda_role_name
policy = jsonencode({ Version = "2012-10-17" Statement = [ { Effect = "Allow" Action = [ "dynamodb:GetItem", "dynamodb:PutItem", "dynamodb:UpdateItem", "dynamodb:DeleteItem", "dynamodb:Query", "dynamodb:Scan", "dynamodb:BatchGetItem", "dynamodb:BatchWriteItem", ] Resource = [ module.my_table.table_arn, "${module.my_table.table_arn}/index/*", ] }, { Effect = "Allow" Action = [ "kms:Encrypt", "kms:Decrypt", "kms:ReEncrypt*", "kms:GenerateDataKey*", "kms:DescribeKey" ] Resource = [module.my_table.kms_key_arn] }, ] })}Deletion Protection
Section titled “Deletion Protection”Deletion protection is enabled by default to prevent accidental table deletion.
Disable Deletion Protection
Section titled “Disable Deletion Protection”Disable it for environments where table deletion is expected, such as short-lived development or preview stacks.
import { MyTable } from ':my-scope/common-constructs';
const table = new MyTable(this, 'Table', { deletionProtection: false,});module "my_table" { source = "../../common/terraform/src/app/dynamodb/my-table" deletion_protection_enabled = false}Billing Mode
Section titled “Billing Mode”The table defaults to on-demand (PAY_PER_REQUEST) billing. Switch to provisioned capacity for predictable, high-throughput workloads.
import { BillingMode } from 'aws-cdk-lib/aws-dynamodb';import { MyTable } from ':my-scope/common-constructs';
const table = new MyTable(this, 'Table', { billingMode: BillingMode.PROVISIONED, readCapacity: 5, writeCapacity: 5,});module "my_table" { source = "../../common/terraform/src/app/dynamodb/my-table" billing_mode = "PROVISIONED"}Point-in-time Recovery
Section titled “Point-in-time Recovery”Point-in-time recovery is enabled by default, allowing you to restore the table to any point in the last 35 days.
Disable Point-in-time Recovery
Section titled “Disable Point-in-time Recovery”import { MyTable } from ':my-scope/common-constructs';
const table = new MyTable(this, 'Table', { pointInTimeRecoverySpecification: { pointInTimeRecoveryEnabled: false },});module "my_table" { source = "../../common/terraform/src/app/dynamodb/my-table" point_in_time_recovery_enabled = false}Encryption Key Rotation
Section titled “Encryption Key Rotation”The KMS key used to encrypt the table has automatic key rotation enabled by default. Disable it if your security policy manages rotation externally.
Disable Encryption Key Rotation
Section titled “Disable Encryption Key Rotation”import { MyTable } from ':my-scope/common-constructs';
const table = new MyTable(this, 'Table', { enableKeyRotation: false,});module "my_table" { source = "../../common/terraform/src/app/dynamodb/my-table" enable_key_rotation = false}Connections
Section titled “Connections”Use the connection generator to integrate this project with others in your workspace. The following connections involve this project: