py#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.

Uso

Gerar um Projeto DynamoDB

VSCode

1. Instale o Nx Console VSCode Plugin se ainda não o fez
2. Abra o console Nx no VSCode
3. Clique em Generate (UI) na seção "Common Nx Commands"
4. Procure por @aws/nx-plugin - py#dynamodb
5. Preencha os parâmetros obrigatórios
6. Clique em Generate

CLI

pnpm

pnpm nx g @aws/nx-plugin:py#dynamodb

yarn

yarn nx g @aws/nx-plugin:py#dynamodb

npm

npx nx g @aws/nx-plugin:py#dynamodb

bun

bunx nx g @aws/nx-plugin:py#dynamodb

Você também pode realizar uma execução simulada para ver quais arquivos seriam alterados

pnpm

pnpm nx g @aws/nx-plugin:py#dynamodb --dry-run

yarn

yarn nx g @aws/nx-plugin:py#dynamodb --dry-run

npm

npx nx g @aws/nx-plugin:py#dynamodb --dry-run

bun

bunx nx g @aws/nx-plugin:py#dynamodb --dry-run

Opções

Parâmetro	Tipo	Padrão	Descrição
name Obrigatório	string	-	Nome do projeto DynamoDB a ser gerado
directory	string	packages	O diretório onde armazenar o projeto.
subDirectory	string	-	O subdiretório onde o projeto é colocado. Por padrão, este é o nome do projeto.
framework	pynamodb	pynamodb	O framework a ser usado para entidades DynamoDB.
tableName	string	-	O nome da tabela DynamoDB. Gerado automaticamente se não especificado.
infra	dynamodb | none	dynamodb	Infraestrutura a ser provisionada para a tabela DynamoDB.
iac	inherit | cdk | terraform	inherit	O provedor IaC preferido. Por padrão, este é herdado da sua seleção inicial.
preferInstallDependencies	boolean	true	Se 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.

Saída do Gerador

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

• Directory<name>

  • __init__.py Exportações do pacote
  • client.py Cliente DynamoDB e resolução de nome da tabela
  • Directoryentities

    • base.py Modelo base PynamoDB com declarações GSI
    • example.py Definição de entidade de exemplo
    • __init__.py Exportações de entidades
• config.json Configuração da tabela incluindo definições GSI e configurações de desenvolvimento local
• project.json Configuração do projeto e alvos de build

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 Cria a tabela DynamoDB na instância local do DynamoDB Local
  • pull-image.ts Baixa a imagem do DynamoDB Local
  • start-container.ts Inicia o contêiner do DynamoDB Local

Infraestrutura

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

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

CDK

• Directorypackages/common/constructs

  • Directorysrc

    • Directoryapp/ Constructs para infraestrutura específica de um projeto/gerador

      • …
    • Directorycore/ Constructs genéricos reutilizados pelos constructs em app

      • …
    • index.ts Ponto de entrada exportando os constructs de app
  • project.json Metas de build e configuração do projeto

Projeto gerado

Este projeto é gerado usando o gerador ts#project e portanto configura as mesmas metas de build.

Terraform

• Directorypackages/common/terraform

  • Directorysrc

    • Directoryapp/ Módulos Terraform para infraestrutura específica de um projeto/gerador

      • …
    • Directorycore/ Módulos genéricos reutilizados pelos módulos em app

      • …
  • project.json Metas de build e configuração do projeto

Projeto gerado

Este projeto é gerado usando o gerador terraform#project e portanto configura as mesmas metas de build.

CDK

• Directorypackages/common/constructs/src

  • Directoryapp

    • Directorydynamodb

      • <name>.ts Infraestrutura específica para sua tabela
  • Directorycore

    • dynamodb.ts Construto genérico de tabela DynamoDB

Terraform

• Directorypackages/common/terraform/src

  • Directoryapp

    • Directorydynamodb

      • Directory<name>

        • <name>.tf Módulo específico para sua tabela
  • Directorycore

    • Directorydynamodb

      • dynamodb.tf Módulo genérico DynamoDB

Desenvolvimento Local

Iniciando o DynamoDB Local

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

pnpm

pnpm nx dev <project-name>

yarn

yarn nx dev <project-name>

npm

npx nx dev <project-name>

bun

bunx nx dev <project-name>

Isso automaticamente:

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

Modelagem de Dados

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 de config.json quando executado localmente via DynamoDB Local. Sem isso, o PynamoDB não saberá qual tabela usar. 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 entidade 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.

Projetando em Torno de Padrões de Acesso

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 categoria — gsi1, pk=CATEGORY#<category>
• Listar por data de criação — gsi2, 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
    """

Armazenando Múltiplos Tipos de Entidades

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.

PynamoDB não foi projetado especificamente para design de tabela única

O DiscriminatorAttribute do PynamoDB foi projetado para polimorfismo de hierarquia de classes dentro de uma única tabela homogênea — não para armazenar tipos de entidades estruturalmente diferentes em uma tabela. O padrão usado aqui funciona, mas tenha essas ressalvas em mente:

• Colisão de chaves é sua responsabilidade. PynamoDB não impede que dois tipos de entidades escrevam o mesmo par pk/sk. Sempre use prefixos de tipo únicos (por exemplo, ORDER#, USER#).
• Consultas entre entidades requerem BaseModel. PynamoDB adiciona automaticamente um filtro discriminador ao consultar uma subclasse específica, então itens de outros tipos de entidades na mesma chave nunca serão retornados. Para recuperar uma mistura de tipos de uma chave de partição GSI compartilhada, sempre consulte via BaseModel.
• Configurações de tabela compartilhadas. Todas as entidades compartilham a configuração Meta (nome da tabela, região, credenciais) definida em BaseModel.

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

Sobrecarga de GSI

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.

• UserModel — gsi2pk=USER, gsi2sk=<created_at> → listar todos os usuários por data
• OrderModel — gsi2pk=ORDER, gsi2sk=<created_at> → listar todos os pedidos por data

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, 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 única requisição, com 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#'),
))

Relacionamentos Um-para-Muitos

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ário — gsi1: pk=USER#<user_id>

Uma alternativa às buscas 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
    """
    ...

# Recuperar o usuário e todos os seus pedidos em uma única consulta à tabela primária
# BaseModel despacha cada item para sua subclasse correta 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)]

O trade-off: 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

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.

Usando o Cliente 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.

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

• Conexão — BaseModel.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, estes são deixados indefinidos para que o PynamoDB use a cadeia de credenciais padrão.
• Nome da tabela — BaseModel._get_connection() chama get_table_name() antes de cada operação, para que a tabela correta seja resolvida em tempo de execução sem qualquer configuração manual.

Parando o DynamoDB Local

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.

Windows

Devido a limitações no tratamento de sinais no Windows, o contêiner não é removido automaticamente quando o dev é interrompido. Você precisará removê-lo manualmente:

<engine> rm -f <scope>-dynamodb

Substitua <engine> pelo seu motor de contêiner (docker ou finch) e <scope> pelo escopo do seu workspace Nx (por exemplo, proj).

Adicionar/Remover Í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 e, em seguida, reflita a mudança em BaseModel adicionando ou removendo a classe GlobalSecondaryIndex correspondente e os 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 local — dev lê config.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

Um GSI por Implantação

Cuidado

O DynamoDB não permite que mais de um GSI seja criado ou excluído em uma única atualização de tabela.

Cada implantação pode adicionar ou remover no máximo um GSI. Se você precisar adicionar ou remover vários GSIs, faça isso um de cada vez — atualize config.json e as classes de entidade relevantes, implante a stack, e então repita para a próxima mudança.

Conectando à Tabela

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

Configuração de runtime

Ao executar na AWS, get_table_name() busca o nome da tabela do AWS AppConfig usando a variável de ambiente RUNTIME_CONFIG_APP_ID. Projetos construídos com este plugin já têm esta variável configurada automaticamente. Para outros projetos Python, certifique-se de que RUNTIME_CONFIG_APP_ID esteja definido no ambiente de runtime com o ID da aplicação AppConfig provisionado pela sua infraestrutura. Para mais informações, consulte o guia de Configuração de Runtime.

Implantando sua Tabela

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

CDK

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

Terraform

O módulo Terraform é criado em common/terraform. Exemplo de uso:

packages/infra/src/main.tf

module "my_table" {
  source = "../../common/terraform/src/app/dynamodb/my-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

Concedendo Acesso

Para permitir que funções Lambda acessem a tabela DynamoDB, conceda as permissões necessárias em sua infraestrutura.

CDK

Chame grantReadWriteData no construto da tabela. Isso concede as permissões do DynamoDB e KMS necessárias para a função de execução do Lambda:

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

Nota

Este exemplo concede acesso a todos os manipuladores em sua API. Se apenas alguns manipuladores precisarem de acesso, configure-os individualmente.

Terraform

Conceda à função de execução do Lambda permissão para acessar a tabela DynamoDB e sua chave de criptografia KMS:

packages/infra/src/main.tf

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

Proteção contra Exclusão

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

Desabilitar Proteção contra Exclusão

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

CDK

packages/infra/src/stacks/application-stack.ts

import { MyTable } from ':my-scope/common-constructs';

const table = new MyTable(this, 'Table', {
  deletionProtection: false,
});

Terraform

packages/infra/src/main.tf

module "my_table" {
  source                      = "../../common/terraform/src/app/dynamodb/my-table"
  deletion_protection_enabled = false
}

Modo de Cobrança

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.

CDK

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

Terraform

packages/infra/src/main.tf

module "my_table" {
  source       = "../../common/terraform/src/app/dynamodb/my-table"
  billing_mode = "PROVISIONED"
}

Recuperação Point-in-time

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

Desabilitar Recuperação Point-in-time

CDK

packages/infra/src/stacks/application-stack.ts

import { MyTable } from ':my-scope/common-constructs';

const table = new MyTable(this, 'Table', {
  pointInTimeRecoverySpecification: { pointInTimeRecoveryEnabled: false },
});

Terraform

packages/infra/src/main.tf

module "my_table" {
  source                         = "../../common/terraform/src/app/dynamodb/my-table"
  point_in_time_recovery_enabled = false
}

Rotação de Chave de Criptografia

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.

Desabilitar Rotação de Chave de Criptografia

CDK

packages/infra/src/stacks/application-stack.ts

import { MyTable } from ':my-scope/common-constructs';

const table = new MyTable(this, 'Table', {
  enableKeyRotation: false,
});

Terraform

packages/infra/src/main.tf

module "my_table" {
  source              = "../../common/terraform/src/app/dynamodb/my-table"
  enable_key_rotation = false
}

Conexões

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

FastAPI para Python DynamoDB Conecte uma FastAPI a uma tabela DynamoDB

Python Agent para Python DynamoDB Conecte um Python Agent a uma tabela DynamoDB

Python MCP Server para Python DynamoDB Conecte um Python MCP Server a uma tabela DynamoDB