py#dynamodb

Este generador crea un nuevo proyecto Python respaldado por Amazon DynamoDB, utilizando PynamoDB para el modelado de entidades. Genera el código de aplicación y la infraestructura necesarios para aprovisionar y administrar una tabla DynamoDB usando AWS CDK o Terraform, con soporte para diseño de tabla única y desarrollo local integrado a través de DynamoDB Local.

Uso

Instale el Nx Console VSCode Plugin si aún no lo ha hecho
Abra la consola Nx en VSCode
Haga clic en Generate (UI) en la sección "Common Nx Commands"
Busque @aws/nx-plugin - py#dynamodb
Complete los parámetros requeridos
Haga clic en Generate

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

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

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

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

También puede realizar una ejecución en seco para ver qué archivos se cambiarían

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

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

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

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

Opciones

Parámetro	Tipo	Predeterminado	Descripción
name Requerido	string	-	Nombre del proyecto DynamoDB a generar
directory	string	packages	El directorio donde almacenar el proyecto.
subDirectory	string	-	El subdirectorio donde se coloca el proyecto. Por defecto es el nombre del proyecto.
framework	pynamodb	pynamodb	El framework a utilizar para las entidades DynamoDB.
tableName	string	-	El nombre de la tabla DynamoDB. Se genera automáticamente si no se especifica.
infra	dynamodb \| none	dynamodb	Infraestructura a aprovisionar para la tabla DynamoDB.
iac	inherit \| cdk \| terraform	inherit	El proveedor IaC preferido. Por defecto se hereda de tu selección inicial.
preferInstallDependencies	boolean	true	Si se prefiere instalar las dependencias después de que se ejecute el generador. Establece en false para diferir la instalación cuando se ejecutan múltiples generadores en lote (la instalación aún se ejecuta si es necesario para que los generadores subsiguientes puedan calcular el grafo de proyectos de Nx); instala una vez al final.

Salida del Generador

El generador crea la siguiente estructura de proyecto en el directorio <directory>/<name>:

Directorio<name>
- __init__.py Exportaciones del paquete
- client.py Cliente DynamoDB y resolución del nombre de tabla
- Directorioentities
  - base.py Modelo base PynamoDB con declaraciones GSI
  - example.py Definición de entidad de ejemplo
  - __init__.py Exportaciones de entidades
config.json Configuración de tabla incluyendo definiciones GSI y configuración de desarrollo local
project.json Configuración del proyecto y objetivos de compilación

Los scripts de desarrollo local se comparten entre todos los proyectos DynamoDB (tanto TypeScript como Python) y se generan una vez en:

Directoriopackages/common/scripts/src/dynamodb
- create-local-table.ts Crea la tabla DynamoDB en la instancia local de DynamoDB Local
- pull-image.ts Descarga la imagen de DynamoDB Local
- start-container.ts Inicia el contenedor de DynamoDB Local

Infraestructura

Dado que este generador proporciona infraestructura como código basada en tu proveedor de iacProvider seleccionado, creará un proyecto en packages/common que incluye los constructos CDK o módulos de Terraform correspondientes.

El proyecto común de infraestructura como código tiene la siguiente estructura:

CDK
Terraform

Directoriopackages/common/constructs
- Directoriosrc
  - Directorioapp/ Constructos para infraestructura específica de un proyecto/generador
    …
  - Directoriocore/ Constructos genéricos reutilizados por los constructos en app
    …
  - index.ts Punto de entrada que exporta los constructos de app
- project.json Objetivos de compilación y configuración del proyecto

Directoriopackages/common/terraform
- Directoriosrc
  - Directorioapp/ Módulos de Terraform para infraestructura específica de un proyecto/generador
    …
  - Directoriocore/ Módulos genéricos reutilizados por los módulos en app
    …
- project.json Objetivos de compilación y configuración del proyecto

CDK
Terraform

Directoriopackages/common/constructs/src
- Directorioapp
  - Directoriodynamodb
    <name>.ts Infraestructura específica para tu tabla
- Directoriocore
  - dynamodb.ts Construcción genérica de tabla DynamoDB

Desarrollo Local

Iniciar DynamoDB Local

El generador configura un target dev que inicia una instancia de DynamoDB Local y crea la tabla. Usa el target dev del proyecto:

pnpm nx dev <project-name>

yarn nx dev <project-name>

npx nx dev <project-name>

bunx nx dev <project-name>

Esto automáticamente:

Descarga la imagen de DynamoDB Local (target pull-image)
Inicia un contenedor
Crea una tabla local con los índices definidos en config.json

Modelado de Datos

El proyecto generado utiliza PynamoDB para el modelado de entidades. Todas las entidades deben heredar del BaseModel generado — este resuelve el nombre correcto de la tabla DynamoDB en tiempo de ejecución, leyendo desde AWS AppConfig cuando está desplegado o desde config.json cuando se ejecuta localmente a través de DynamoDB Local. Sin esto, PynamoDB no sabrá qué tabla usar. BaseModel también utiliza el soporte de polimorfismo de PynamoDB para almacenar múltiples tipos de entidades en una sola tabla, siguiendo el diseño de tabla única de DynamoDB.

Agrega o actualiza archivos de entidades bajo <name>/entities/, utilizando la entidad de ejemplo generada como punto de partida:

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 más detalles, consulta el tutorial de PynamoDB.

Diseñar en Torno a Patrones de Acceso

En DynamoDB, el diseño del esquema comienza con tus consultas, no con la forma de tus datos. Antes de escribir cualquier modelo, enumera cada patrón de acceso que tu aplicación necesita, luego diseña los valores de clave pk, sk y GSI para que cada patrón sea respondido por una sola solicitud de tabla — sin JOINs, sin lecturas secuenciales.

El ExampleModel generado demuestra esto para tres patrones:

Obtener por ID — índice primario, pk=EXAMPLE#<id>, sk=EXAMPLE#<id>
Listar por categoría — gsi1, pk=CATEGORY#<category>
Listar por fecha de creación — gsi2, pk=EXAMPLE, clave de ordenamiento entre marcas de tiempo ISO

La convención de prefijo de tipo (por ejemplo, EXAMPLE#, CATEGORY#) es deliberada: hace que los elementos sean autodescriptivos al navegar por la tabla, previene colisiones accidentales de claves entre tipos de entidades que comparten un índice, y permite el filtrado de prefijos de clave de ordenamiento usando begins_with.

Antes de escribir una nueva entidad, define sus patrones de clave por adelantado en un docstring. El OrderModel en la siguiente sección sigue esta convención:

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

Almacenar Múltiples Tipos de Entidades

El DiscriminatorAttribute de PynamoDB almacena una etiqueta de tipo (entity_type) en cada elemento. Al consultar a través de BaseModel, esta etiqueta se utiliza para instanciar cada resultado como su subclase correcta automáticamente — por lo que una sola consulta puede devolver una mezcla de UserModel, OrderModel y cualquier otro tipo de entidad registrado en la misma tabla.

El DiscriminatorAttribute de PynamoDB fue diseñado para polimorfismo de jerarquía de clases dentro de una sola tabla homogénea — no para almacenar tipos de entidades estructuralmente diferentes en una tabla. El patrón utilizado aquí funciona, pero ten en cuenta estas advertencias:

La colisión de claves es tu responsabilidad. PynamoDB no previene que dos tipos de entidades escriban el mismo par pk/sk. Siempre usa prefijos de tipo únicos (por ejemplo, ORDER#, USER#).
Las consultas entre entidades requieren BaseModel. PynamoDB agrega automáticamente un filtro discriminador al consultar una subclase específica, por lo que los elementos de otros tipos de entidades en la misma clave nunca serán devueltos. Para recuperar una mezcla de tipos de una clave de partición GSI compartida, siempre consulta a través de BaseModel.
Configuración de tabla compartida. Todas las entidades comparten la configuración Meta (nombre de tabla, región, credenciales) definida en BaseModel.

A continuación se muestra un ejemplo completo de dos entidades — un UserModel con registros OrderModel asociados almacenados en la misma tabla:

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)

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)

Exporta las nuevas entidades desde __init__.py:

from .user import UserModel
from .order import OrderModel
from .example import ExampleModel

Sobrecarga de GSI

BaseModel proporciona dos GSI compartidos (gsi1_index, gsi2_index). Tanto UserModel como OrderModel arriba escriben en gsi2 — pero con diferentes valores de gsi2pk (USER vs ORDER). Esto es sobrecarga de GSI: reutilizar un solo índice físico para servir múltiples patrones de acceso independientes sin consumir capacidad GSI adicional.

UserModel — gsi2pk=USER, gsi2sk=<created_at> → listar todos los usuarios por fecha
OrderModel — gsi2pk=ORDER, gsi2sk=<created_at> → listar todos los pedidos por fecha

gsi1 también puede sobrecargarse cuando múltiples tipos de entidades comparten el mismo padre. Si más tarde agregas un ReviewModel que también pertenece a un usuario, puedes asignarle gsi1pk=USER#<user_id> con una clave de ordenamiento REVIEW#<id> — no se necesita GSI adicional. Consultar gsi1 a través de BaseModel entonces devuelve tanto pedidos como reseñas para ese usuario en una solicitud, con PynamoDB instanciando cada elemento como su subclase correcta:

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 solo un tipo de entidad de un GSI sobrecargado, usa una condición de prefijo de clave de ordenamiento:

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

Relaciones Uno a Muchos

En una relación uno a muchos la entidad hija almacena una referencia a su padre en una clave de partición GSI, haciendo que la relación sea transitable en ambas direcciones sin duplicar datos. El ejemplo UserModel / OrderModel arriba es exactamente este patrón:

Obtener un solo pedido por ID — tabla primaria: pk=ORDER#<id>, sk=ORDER#<id>
Listar todos los pedidos de un usuario — gsi1: pk=USER#<user_id>

Una alternativa a las búsquedas basadas en GSI es el patrón de colección de elementos: dar a los elementos hijos el mismo pk que su padre y usar la clave de ordenamiento para diferenciarlos. Esto te permite recuperar el padre y todos sus hijos en una sola consulta de tabla primaria, sin un GSI:

class OrderModel(BaseModel, discriminator='OrderModel'):
    """
    Key design (item collection):
      pk=USER#<user_id>,    sk=ORDER#<order_id>   <- co-located under the parent user
    """
    ...

# Recuperar el usuario y todos sus pedidos en una consulta de tabla primaria
# BaseModel despacha cada elemento a su subclase correcta a través de 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)]

El compromiso: las colecciones de elementos colocan todos los hijos bajo una sola clave de partición, lo cual es óptimo para la mayoría de las cargas de trabajo pero puede crear una partición caliente con un rendimiento de escritura extremo. El enfoque GSI (usado en los ejemplos anteriores) mantiene cada entidad en su propia partición y generalmente es más seguro para comenzar.

Relaciones Muchos a Muchos

Las relaciones muchos a muchos requieren una entidad de unión usando el patrón de lista de adyacencia: un elemento dedicado que registra cada enlace, con su clave GSI invirtiendo la dirección para que la relación pueda ser atravesada en ambos sentidos.

Considera ArticleModel y TagModel, donde un artículo puede tener muchas etiquetas y una etiqueta puede aplicarse a muchos artículos:

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

Debido a que ArticleTagModel usa pk=ARTICLE#<article_id> — la misma partición que el artículo mismo — puedes recuperar un artículo y todas sus etiquetas en una sola consulta de tabla primaria:

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 más información sobre el modelado de datos en DynamoDB, consulta la guía de modelado de datos de DynamoDB y Creación de un diseño de tabla única con Amazon DynamoDB.

Usar el Cliente DynamoDB

El client.py generado exporta dos utilidades clave:

is_local() — devuelve True cuando LOCAL_DEV=true, utilizado para cambiar entre comportamiento local y AWS.
get_table_name() — devuelve el nombre de la tabla DynamoDB. Cuando LOCAL_DEV=true, lee el nombre de la tabla desde localDev.tableName en config.json; de lo contrario, obtiene el nombre desde AWS AppConfig usando la variable de entorno RUNTIME_CONFIG_APP_ID y lo almacena en caché para llamadas posteriores.

BaseModel en entities/base.py utiliza ambos para configurar PynamoDB automáticamente:

Conexión — BaseModel.Meta establece host desde config.json y codifica region, aws_access_key_id y aws_secret_access_key cuando is_local() es True, apuntando PynamoDB a la instancia local de DynamoDB. En AWS, estos se dejan sin configurar para que PynamoDB use la cadena de credenciales predeterminada.
Nombre de tabla — BaseModel._get_connection() llama a get_table_name() antes de cada operación, por lo que la tabla correcta se resuelve en tiempo de ejecución sin ninguna configuración manual.

Detener DynamoDB Local

Detener dev (por ejemplo, con Ctrl+C) elimina automáticamente el contenedor de DynamoDB Local, pero conserva el volumen nombrado para que tus datos persistan entre reinicios.

Debido a limitaciones con el manejo de señales en Windows, el contenedor no se elimina automáticamente cuando se detiene dev. Necesitarás eliminarlo manualmente:

<engine> rm -f <scope>-dynamodb

Reemplaza <engine> con tu motor de contenedores (docker o finch) y <scope> con el scope de tu espacio de trabajo Nx (por ejemplo, proj).

Agregar/Eliminar Índices Secundarios Globales

Los GSI se definen en config.json en la raíz del proyecto bajo la clave tableConfig.globalSecondaryIndexes. Agrega una entrada para cada GSI, luego refleja el cambio en BaseModel agregando o eliminando la clase GlobalSecondaryIndex correspondiente y los atributos en <name>/entities/base.py:

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

El campo sortKey es opcional para GSIs de solo clave hash.

Este archivo de configuración es la única fuente de verdad leída por todos los consumidores:

Desarrollo local — dev lee config.json y crea o actualiza la tabla local para que coincida con la lista de GSI
CDK — el constructo lee config.json en el momento de síntesis, por lo que los cambios de GSI se reflejan en el próximo cdk deploy
Terraform — el módulo lee config.json en el momento de plan/apply

Un GSI por Despliegue

Conectarse a la Tabla

En cualquier proyecto Python, agrega el paquete DynamoDB como una dependencia del espacio de trabajo e importa las clases de entidad directamente:

from my_db_package.entities import ExampleModel

item = ExampleModel.get_by_id('123')

Desplegar tu Tabla

El generador de DynamoDB crea infraestructura CDK o Terraform basada en tu iac seleccionado.

CDK
Terraform

El constructo CDK se crea en common/constructs. Ejemplo de uso:

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

Esto aprovisiona una tabla DynamoDB con:

pk (clave de partición) y sk (clave de ordenación), ambas de tipo String
Índices Secundarios Globales según se definen en config.json
Facturación bajo demanda (PAY_PER_REQUEST)
Cifrado KMS gestionado por el cliente con rotación automática de claves
Recuperación point-in-time habilitada
Protección contra eliminación habilitada
Nombre de tabla registrado en Runtime Config bajo el namespace dynamodb en AWS AppConfig

El módulo Terraform se crea en common/terraform. Ejemplo de uso:

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

Esto aprovisiona una tabla DynamoDB con:

pk (clave de partición) y sk (clave de ordenación), ambas de tipo String
Índices Secundarios Globales según se definen en config.json
Facturación bajo demanda (PAY_PER_REQUEST)
Cifrado KMS gestionado por el cliente con rotación automática de claves
Recuperación point-in-time habilitada
Protección contra eliminación habilitada
Nombre de tabla registrado en Runtime Config

Concediendo Acceso

Para permitir que las funciones Lambda accedan a la tabla de DynamoDB, otorga los permisos necesarios en tu infraestructura.

CDK
Terraform

Llama a grantReadWriteData en el constructo de la tabla. Esto otorga tanto los permisos de DynamoDB como de KMS requeridos por el rol de ejecución de Lambda:

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

Otorga al rol de ejecución de Lambda permiso para acceder a la tabla de DynamoDB y su clave de cifrado KMS:

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

Protección contra Eliminación

La protección contra eliminación está habilitada por defecto para prevenir la eliminación accidental de la tabla.

Deshabilitar la Protección contra Eliminación

Deshabilítala para entornos donde se espera la eliminación de la tabla, como stacks de desarrollo o preview de corta duración.

CDK
Terraform

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
}

Modo de Facturación

La tabla utiliza por defecto facturación bajo demanda (PAY_PER_REQUEST). Cambia a capacidad aprovisionada para cargas de trabajo predecibles de alto rendimiento.

CDK
Terraform

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

Recuperación Point-in-time

La recuperación point-in-time está habilitada por defecto, permitiéndote restaurar la tabla a cualquier punto en los últimos 35 días.

Deshabilitar la Recuperación Point-in-time

CDK
Terraform

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
}

Rotación de Clave de Cifrado

La clave KMS utilizada para cifrar la tabla tiene la rotación automática de claves habilitada por defecto. Deshabilítala si tu política de seguridad gestiona la rotación externamente.

Deshabilitar la Rotación de Clave de Cifrado

CDK
Terraform

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
}

Conexiones

Usa el generador connection para integrar este proyecto con otros en tu workspace. Las siguientes conexiones involucran este proyecto:

FastAPI a Python DynamoDB Conecta una FastAPI a una tabla DynamoDB

Python Agent a Python DynamoDB Conecta un Python Agent a una tabla DynamoDB

Python MCP Server a Python DynamoDB Conecta un Python MCP Server a una tabla DynamoDB