Skip to content

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.

  1. Install the Nx Console VSCode Plugin if you haven't already
  2. Open the Nx Console in VSCode
  3. Click Generate (UI) in the "Common Nx Commands" section
  4. Search for @aws/nx-plugin - py#dynamodb
  5. Fill in the required parameters
    • Click Generate
    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.

    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

    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/constructs/src
      • Directoryapp
        • Directorydynamodb
          • <name>.ts Infrastructure specific to your table
      • Directorycore
        • dynamodb.ts Generic DynamoDB table construct

    The generator configures a dev target that starts a DynamoDB Local instance and creates the table. Use the project’s dev target:

    Terminal window
    pnpm nx dev <project-name>

    This automatically:

    1. Pulls the DynamoDB Local image (pull-image target)
    2. Starts a container
    3. Creates a local table with the indexes defined in config.json

    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:

    packages/my_table/my_table/entities/example.py
    from collections.abc import Iterator
    from datetime import UTC, datetime
    from pynamodb.attributes import UnicodeAttribute
    from .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.

    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 categorygsi1, pk=CATEGORY#<category>
    • List by creation dategsi2, 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
    """

    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:

    packages/my_table/my_table/entities/user.py
    from collections.abc import Iterator
    from datetime import UTC, datetime
    from pynamodb.attributes import UnicodeAttribute
    from .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)
    packages/my_table/my_table/entities/order.py
    from collections.abc import Iterator
    from datetime import UTC, datetime
    from pynamodb.attributes import UnicodeAttribute
    from .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:

    packages/my_table/my_table/entities/__init__.py
    from .user import UserModel
    from .order import OrderModel
    from .example import ExampleModel

    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.

    • UserModelgsi2pk=USER, gsi2sk=<created_at> → list all users by date
    • OrderModelgsi2pk=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 BaseModel
    from .order import OrderModel
    from .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#'),
    ))

    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 usergsi1: 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:

    packages/my_table/my_table/entities/order.py (item collection variant)
    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 DiscriminatorAttribute
    items = 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 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:

    packages/my_table/my_table/entities/article_tag.py
    from collections.abc import Iterator
    from pynamodb.attributes import UnicodeAttribute
    from .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 BaseModel
    from .article import ArticleModel
    from .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.

    The generated client.py exports two key utilities:

    • is_local() — returns True when LOCAL_DEV=true, used to switch between local and AWS behaviour.
    • get_table_name() — returns the DynamoDB table name. When LOCAL_DEV=true, reads the table name from localDev.tableName in config.json; otherwise fetches the name from AWS AppConfig using the RUNTIME_CONFIG_APP_ID environment variable and caches it for subsequent calls.

    BaseModel in entities/base.py uses both to configure PynamoDB automatically:

    • ConnectionBaseModel.Meta sets host from config.json and hardcodes region, aws_access_key_id, and aws_secret_access_key when is_local() is True, pointing PynamoDB at the local DynamoDB instance. In AWS, these are left unset so PynamoDB uses the default credential chain.
    • Table nameBaseModel._get_connection() calls get_table_name() before each operation, so the correct table is resolved at runtime without any manual configuration.

    Stopping dev (e.g. with Ctrl+C) automatically removes the DynamoDB Local container, but preserves the named volume so your data persists across restarts.

    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:

    config.json
    {
    ...
    "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 developmentdev reads config.json and creates or updates the local table to match the GSI list
    • CDK — the construct reads config.json at synth time, so GSI changes are reflected on the next cdk deploy
    • Terraform — the module reads config.json at plan/apply time

    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')

    The DynamoDB generator creates CDK or Terraform infrastructure based on your selected iac.

    The CDK construct is created in common/constructs. Example usage:

    packages/infra/src/stacks/application-stack.ts
    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) and sk (sort key), both String type
    • 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 dynamodb namespace in AWS AppConfig

    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:

    packages/infra/src/stacks/application-stack.ts
    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);
    });

    Deletion protection is enabled by default to prevent accidental table deletion.

    Disable it for environments where table deletion is expected, such as short-lived development or preview stacks.

    packages/infra/src/stacks/application-stack.ts
    import { MyTable } from ':my-scope/common-constructs';
    const table = new MyTable(this, 'Table', {
    deletionProtection: false,
    });

    The table defaults to on-demand (PAY_PER_REQUEST) billing. Switch to provisioned capacity for predictable, high-throughput workloads.

    packages/infra/src/stacks/application-stack.ts
    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,
    });

    Point-in-time recovery is enabled by default, allowing you to restore the table to any point in the last 35 days.

    packages/infra/src/stacks/application-stack.ts
    import { MyTable } from ':my-scope/common-constructs';
    const table = new MyTable(this, 'Table', {
    pointInTimeRecoverySpecification: { pointInTimeRecoveryEnabled: false },
    });

    The KMS key used to encrypt the table has automatic key rotation enabled by default. Disable it if your security policy manages rotation externally.

    packages/infra/src/stacks/application-stack.ts
    import { MyTable } from ':my-scope/common-constructs';
    const table = new MyTable(this, 'Table', {
    enableKeyRotation: false,
    });

    Use the connection generator to integrate this project with others in your workspace. The following connections involve this project:

    FastAPI Amazon DynamoDB Python
    FastAPI to Python DynamoDB Connect a FastAPI to a DynamoDB table
    Strands Agents Python Amazon DynamoDB Python
    Python Agent to Python DynamoDB Connect a Python Agent to a DynamoDB table
    Model Context Protocol Python Amazon DynamoDB Python
    Python MCP Server to Python DynamoDB Connect a Python MCP Server to a DynamoDB table