Pular para o conteúdo

Python DynamoDB

Este gerador cria um novo projeto Python apoiado pelo Amazon DynamoDB, usando PynamoDB para modelagem de entidades. Ele gera o código da aplicação e a infraestrutura necessária para provisionar e gerenciar uma tabela DynamoDB usando AWS CDK ou Terraform, com suporte a design de tabela única e desenvolvimento local integrado via DynamoDB Local.

Terminal window
pnpm nx g @aws/nx-plugin:py#dynamodb
Você também pode realizar uma execução simulada para ver quais arquivos seriam alterados
Terminal window
pnpm nx g @aws/nx-plugin:py#dynamodb --dry-run
ParâmetroTipoPadrãoDescrição
name Obrigatóriostring-Nome do projeto DynamoDB a ser gerado
directory stringpackagesO diretório onde armazenar o projeto.
subDirectory string-O subdiretório onde o projeto é colocado. Por padrão, este é o nome do projeto.
framework pynamodbpynamodbO framework a ser usado para entidades DynamoDB.
tableName string-O nome da tabela DynamoDB. Gerado automaticamente se não especificado.
infra dynamodb | nonedynamodbInfraestrutura a ser provisionada para a tabela DynamoDB.
iac inherit | cdk | terraforminheritO provedor IaC preferido. Por padrão, este é herdado da sua seleção inicial.
preferInstallDependencies booleantrueSe deve preferir instalar dependências após a execução do gerador. Defina como false para adiar a instalação ao executar múltiplos geradores em lote (uma instalação ainda é executada se necessário para que geradores subsequentes possam calcular o grafo de projetos Nx); instale uma vez no final.

O gerador cria a seguinte estrutura de projeto no diretório <directory>/<name>:

  • 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

Os scripts de desenvolvimento local são compartilhados entre todos os projetos DynamoDB (tanto TypeScript quanto Python) e gerados uma vez em:

  • 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

Como este gerador fornece infraestrutura como código baseada no seu iac escolhido, ele criará um projeto em packages/common que inclui os constructs CDK relevantes ou módulos Terraform.

O projeto comum de infraestrutura como código é estruturado da seguinte forma:

  • 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 Infraestrutura específica para sua tabela
    • Directorycore
      • dynamodb.ts Construto genérico de tabela DynamoDB

O gerador configura um target dev que inicia uma instância do DynamoDB Local e cria a tabela. Use o target dev do projeto:

Terminal window
pnpm nx dev <project-name>

Isso automaticamente:

  1. Baixa a imagem do DynamoDB Local (target pull-image)
  2. Inicia um container
  3. Cria uma tabela local com os índices definidos em config.json

O projeto gerado usa PynamoDB para modelagem de entidades. Todas as entidades devem herdar do BaseModel gerado — ele resolve o nome correto da tabela DynamoDB em tempo de execução, lendo do AWS AppConfig quando implantado ou do config.json quando executado localmente via DynamoDB Local. Sem isso, o PynamoDB não saberá qual tabela usar. O BaseModel também usa suporte a polimorfismo do PynamoDB para armazenar múltiplos tipos de entidades em uma única tabela, seguindo o design de tabela única do DynamoDB.

Adicione ou atualize arquivos de entidades em <name>/entities/, usando a entidade de exemplo gerada como ponto de partida:

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

Para mais detalhes, consulte o tutorial do PynamoDB.

No DynamoDB, o design de esquema começa com suas consultas, não com a forma dos seus dados. Antes de escrever qualquer modelo, liste todos os padrões de acesso que sua aplicação precisa, depois projete os valores de chave pk, sk e GSI para que cada padrão seja respondido por uma única requisição à tabela — sem JOINs, sem leituras sequenciais.

O ExampleModel gerado demonstra isso para três padrões:

  • Obter por ID — índice primário, pk=EXAMPLE#<id>, sk=EXAMPLE#<id>
  • Listar por categoriagsi1, pk=CATEGORY#<category>
  • Listar por data de criaçãogsi2, pk=EXAMPLE, chave de ordenação entre timestamps ISO

A convenção de prefixo de tipo (por exemplo, EXAMPLE#, CATEGORY#) é deliberada: torna os itens autodescritivos ao navegar pela tabela, previne colisões acidentais de chaves entre tipos de entidades que compartilham um índice, e permite filtragem de prefixo de chave de ordenação usando begins_with.

Antes de escrever uma nova entidade, defina seus padrões de chave antecipadamente em uma docstring. O OrderModel na próxima seção segue essa convenção:

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
"""

O DiscriminatorAttribute do PynamoDB armazena um rótulo de tipo (entity_type) em cada item. Ao consultar via BaseModel, esse rótulo é usado para instanciar cada resultado como sua subclasse correta automaticamente — então uma única consulta pode retornar uma mistura de UserModel, OrderModel e qualquer outro tipo de entidade registrado na mesma tabela.

Abaixo está um exemplo completo de duas entidades — um UserModel com registros OrderModel associados armazenados na mesma tabela:

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)

Exporte as novas entidades de __init__.py:

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

O BaseModel fornece dois GSIs compartilhados (gsi1_index, gsi2_index). Tanto UserModel quanto OrderModel acima escrevem em gsi2 — mas com valores gsi2pk diferentes (USER vs ORDER). Isso é sobrecarga de GSI: reutilizar um único índice físico para servir múltiplos padrões de acesso independentes sem consumir capacidade extra de GSI.

  • UserModelgsi2pk=USER, gsi2sk=<created_at> → listar todos os usuários por data
  • OrderModelgsi2pk=ORDER, gsi2sk=<created_at> → listar todos os pedidos por data

O gsi1 também pode ser sobrecarregado quando múltiplos tipos de entidades compartilham o mesmo pai. Se você adicionar posteriormente um ReviewModel que também pertence a um usuário, você pode atribuir a ele gsi1pk=USER#<user_id> com uma chave de ordenação REVIEW#<id> — nenhum GSI adicional necessário. Consultar gsi1 via BaseModel então retorna tanto pedidos quanto avaliações para aquele usuário em uma requisição, com o PynamoDB instanciando cada item como sua subclasse correta:

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

Para recuperar apenas um tipo de entidade de um GSI sobrecarregado, use uma condição de prefixo de chave de ordenação:

orders_only = list(BaseModel.gsi1_index.query(
f'USER#{user_id}',
range_key_condition=BaseModel.gsi1sk.startswith('ORDER#'),
))

Em um relacionamento um-para-muitos a entidade filha armazena uma referência ao seu pai em uma chave de partição GSI, tornando o relacionamento navegável em ambas as direções sem duplicar dados. O exemplo UserModel / OrderModel acima é exatamente esse padrão:

  • Obter um único pedido por ID — tabela primária: pk=ORDER#<id>, sk=ORDER#<id>
  • Listar todos os pedidos de um usuáriogsi1: pk=USER#<user_id>

Uma alternativa às consultas baseadas em GSI é o padrão de coleção de itens: dê aos itens filhos o mesmo pk que seu pai e use a chave de ordenação para diferenciá-los. Isso permite recuperar o pai e todos os seus filhos em uma única consulta à tabela primária, sem um GSI:

packages/my_table/my_table/entities/order.py (variante de coleção de itens)
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)]

A compensação: coleções de itens colocam todos os filhos sob uma única chave de partição, o que é ideal para a maioria das cargas de trabalho, mas pode criar uma partição quente em throughput de escrita extremo. A abordagem GSI (usada nos exemplos acima) mantém cada entidade em sua própria partição e geralmente é mais segura para começar.

Relacionamentos muitos-para-muitos requerem uma entidade de junção usando o padrão de lista de adjacência: um item dedicado que registra cada link, com sua chave GSI invertendo a direção para que o relacionamento possa ser percorrido em ambos os sentidos.

Considere ArticleModel e TagModel, onde um artigo pode ter muitas tags e uma tag pode se aplicar a muitos artigos:

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

Como ArticleTagModel usa pk=ARTICLE#<article_id> — a mesma partição que o próprio artigo — você pode recuperar um artigo e todas as suas tags em uma única consulta à tabela primária:

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

Para leitura adicional sobre modelagem de dados do DynamoDB, consulte o guia de modelagem de dados do DynamoDB e Criando um design de tabela única com Amazon DynamoDB.

O client.py gerado exporta dois utilitários principais:

  • is_local() — retorna True quando LOCAL_DEV=true, usado para alternar entre comportamento local e AWS.
  • get_table_name() — retorna o nome da tabela DynamoDB. Quando LOCAL_DEV=true, lê o nome da tabela de localDev.tableName em config.json; caso contrário, busca o nome do AWS AppConfig usando a variável de ambiente RUNTIME_CONFIG_APP_ID e o armazena em cache para chamadas subsequentes.

O BaseModel em entities/base.py usa ambos para configurar o PynamoDB automaticamente:

  • ConexãoBaseModel.Meta define host de config.json e codifica region, aws_access_key_id e aws_secret_access_key quando is_local() é True, apontando o PynamoDB para a instância local do DynamoDB. Na AWS, esses são deixados indefinidos para que o PynamoDB use a cadeia de credenciais padrão.
  • Nome da tabelaBaseModel._get_connection() chama get_table_name() antes de cada operação, então a tabela correta é resolvida em tempo de execução sem qualquer configuração manual.

Parar o dev (por exemplo, com Ctrl+C) remove automaticamente o contêiner DynamoDB Local, mas preserva o volume nomeado para que seus dados persistam entre reinicializações.

Adicionando/Removendo Índices Secundários Globais

Seção intitulada “Adicionando/Removendo Índices Secundários Globais”

GSIs são definidos em config.json na raiz do projeto sob a chave tableConfig.globalSecondaryIndexes. Adicione uma entrada para cada GSI, depois reflita a mudança em BaseModel adicionando ou removendo a classe GlobalSecondaryIndex correspondente e atributos em <name>/entities/base.py:

config.json
{
...
"tableConfig": {
"globalSecondaryIndexes": [
{
"indexName": "gsi1pk-gsi1sk-index",
"partitionKey": "gsi1pk",
"sortKey": "gsi1sk"
},
{
"indexName": "gsi2pk-gsi2sk-index",
"partitionKey": "gsi2pk",
"sortKey": "gsi2sk"
}
]
}
}

O campo sortKey é opcional para GSIs somente com chave de hash.

Este arquivo de configuração é a única fonte de verdade lida por todos os consumidores:

  • Desenvolvimento localdevconfig.json e cria ou atualiza a tabela local para corresponder à lista de GSI
  • CDK — o construct lê config.json no momento da síntese, então as mudanças de GSI são refletidas no próximo cdk deploy
  • Terraform — o módulo lê config.json no momento do plan/apply

Em qualquer projeto Python, adicione o pacote DynamoDB como uma dependência do workspace e importe as classes de entidade diretamente:

from my_db_package.entities import ExampleModel
item = ExampleModel.get_by_id('123')

Nos bastidores, ExampleModel chama get_table_name() para buscar o nome da tabela do AWS AppConfig em tempo de execução.

O gerador DynamoDB cria infraestrutura CDK ou Terraform com base no seu iac selecionado.

O construtor CDK é criado em common/constructs. Exemplo de uso:

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

Isso provisiona uma tabela DynamoDB com:

  • pk (chave de partição) e sk (chave de ordenação), ambas do tipo String
  • Índices Secundários Globais conforme definido em config.json
  • Cobrança sob demanda (PAY_PER_REQUEST)
  • Criptografia KMS gerenciada pelo cliente com rotação automática de chaves
  • Recuperação point-in-time habilitada
  • Proteção contra exclusão habilitada
  • Nome da tabela registrado no Runtime Config sob o namespace dynamodb no AWS AppConfig

A proteção contra exclusão está habilitada por padrão para prevenir a exclusão acidental da tabela.

Desabilite-a para ambientes onde a exclusão da tabela é esperada, como stacks de desenvolvimento ou preview de curta duração.

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

A tabela usa por padrão cobrança sob demanda (PAY_PER_REQUEST). Mude para capacidade provisionada para cargas de trabalho previsíveis e de alto throughput.

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,
});

Recuperação point-in-time está habilitada por padrão, permitindo que você restaure a tabela para qualquer ponto nos últimos 35 dias.

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

A chave KMS usada para criptografar a tabela tem rotação automática de chaves habilitada por padrão. Desabilite-a se sua política de segurança gerencia a rotação externamente.

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

Use o gerador connection para integrar este projeto com outros em seu workspace. As seguintes conexões envolvem este projeto:

FastAPIAmazon DynamoDBPython
FastAPI to Python DynamoDBConnect a FastAPI to a DynamoDB table
Strands AgentsPythonAmazon DynamoDBPython
Python Agent to Python DynamoDBConnect a Python Agent to a DynamoDB table
Model Context ProtocolPythonAmazon DynamoDBPython
Python MCP Server to Python DynamoDBConnect a Python MCP Server to a DynamoDB table