Python DynamoDB

Ce générateur crée un nouveau projet Python soutenu par Amazon DynamoDB, utilisant PynamoDB pour la modélisation des entités. Il génère le code d’application et l’infrastructure nécessaires pour provisionner et gérer une table DynamoDB en utilisant AWS CDK ou Terraform, avec prise en charge de la conception à table unique et développement local intégré via DynamoDB Local.

Utilisation

Générer un projet DynamoDB

Exécuter ce générateur@aws/nx-plugin:py#dynamodb

CLI
VSCode

$ 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

Installez le Nx Console VSCode Plugin si ce n'est pas déjà fait
Ouvrez la console Nx dans VSCode
Cliquez sur Generate (UI) dans la section "Common Nx Commands"
Recherchez @aws/nx-plugin - py#dynamodb
Remplissez les paramètres requis
Cliquez sur Generate

Composez votre commande8

nameRequis

Options

Options du générateur8 options

nameRequisstring: Nom du projet DynamoDB à générer
directorystringPar défaut: packages: Le répertoire dans lequel stocker le projet.
frameworkenumPar défaut: pynamodb: Le framework à utiliser pour les entités DynamoDB.
pynamodb
infraenumPar défaut: dynamodb: Infrastructure à provisionner pour la table DynamoDB.
dynamodbnone
iacenumPar défaut: inherit: Le fournisseur IaC préféré. Par défaut, il est hérité de votre sélection initiale.
inheritcdkterraform
subDirectorystring: Le sous-répertoire dans lequel le projet est placé. Par défaut, il s'agit du nom du projet.
tableNamestring: Le nom de la table DynamoDB. Généré automatiquement s'il n'est pas spécifié.
preferInstallDependenciesbooleanPar défaut: true: Indique s'il faut privilégier l'installation des dépendances après l'exécution du générateur. Définir sur false pour différer l'installation lors de l'exécution de plusieurs générateurs en lot (une installation s'exécute quand même si nécessaire pour que les générateurs suivants puissent calculer le graphe de projet Nx) ; installer une seule fois à la fin.

Sortie du générateur

Le générateur crée la structure de projet suivante dans le répertoire <directory>/<name> :

Répertoire<name>
- __init__.py Package exports
- client.py DynamoDB client and table name resolution
- Répertoireentities
  - 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

Les scripts de développement local sont partagés entre tous les projets DynamoDB (TypeScript et Python) et générés une fois dans :

Répertoirepackages/common/scripts/src/dynamodb
- create-local-table.ts Creates the DynamoDB table in the local DynamoDB Local instance
- pull-image.ts Pulls the DynamoDB Local image
- start-container.ts Starts the DynamoDB Local container

Infrastructure

Étant donné que ce générateur fournit de l’infrastructure en tant que code basée sur votre iac choisi, il créera un projet dans packages/common qui inclut les constructs CDK ou modules Terraform pertinents.

Le projet d’infrastructure en tant que code commun est structuré comme suit :

CDK
Terraform

Répertoirepackages/common/constructs
- Répertoiresrc
  - Répertoireapp/ Constructs pour l’infrastructure spécifique à un projet/générateur
    …
  - Répertoirecore/ Constructs génériques qui sont réutilisés par les constructs dans app
    …
  - index.ts Entry point exporting constructs from app
- project.json Project build targets and configuration

Répertoirepackages/common/terraform
- Répertoiresrc
  - Répertoireapp/ Terraform modules for infrastructure specific to a project/generator
    …
  - Répertoirecore/ Generic modules which are reused by modules in app
    …
- project.json Project build targets and configuration

CDK
Terraform

Répertoirepackages/common/constructs/src
- Répertoireapp
  - Répertoiredynamodb
    <name>.ts Infrastructure spécifique à votre table
- Répertoirecore
  - dynamodb.ts Construct générique de table DynamoDB

Architecture

Le projet déployé provisionne la table elle-même, que tout projet auquel il est connecté lit et écrit :

Loading the diagram…

Développement local

Démarrer DynamoDB local

Le générateur configure une cible dev qui démarre une instance DynamoDB Local et crée la table. Utilisez la cible dev du projet :

pnpm nx dev <project-name>

yarn nx dev <project-name>

npx nx dev <project-name>

bunx nx dev <project-name>

Cela effectue automatiquement :

Télécharge l’image DynamoDB Local (cible pull-image)
Démarre un conteneur
Crée une table locale avec les index définis dans config.json

Modélisation des données

Le projet généré utilise PynamoDB pour la modélisation des entités. Toutes les entités doivent hériter du BaseModel généré — il résout le nom de table DynamoDB correct au moment de l’exécution, en lisant depuis AWS AppConfig lorsqu’il est déployé ou depuis config.json lors de l’exécution locale via DynamoDB Local. Sans cela, PynamoDB ne saura pas quelle table utiliser. BaseModel utilise également la prise en charge du polymorphisme de PynamoDB pour stocker plusieurs types d’entités dans une seule table, suivant la conception à table unique de DynamoDB.

Ajoutez ou mettez à jour les fichiers d’entités sous <name>/entities/, en utilisant l’exemple d’entité généré comme point de départ :

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

Pour plus de détails, consultez le tutoriel PynamoDB.

Concevoir autour des modèles d’accès

Dans DynamoDB, la conception du schéma commence par vos requêtes, pas par la forme de vos données. Avant d’écrire un modèle, listez chaque modèle d’accès dont votre application a besoin, puis concevez les valeurs de clés pk, sk et GSI de sorte que chaque modèle soit répondu par une seule requête de table — pas de JOINs, pas de lectures séquentielles.

Le ExampleModel généré démontre cela pour trois modèles :

Obtenir par ID — index primaire, pk=EXAMPLE#<id>, sk=EXAMPLE#<id>
Lister par catégorie — gsi1, pk=CATEGORY#<category>
Lister par date de création — gsi2, pk=EXAMPLE, clé de tri entre horodatages ISO

La convention de préfixe de type (par exemple EXAMPLE#, CATEGORY#) est délibérée : elle rend les éléments auto-descriptifs lors de la navigation dans la table, empêche les collisions de clés accidentelles entre les types d’entités qui partagent un index, et permet le filtrage de préfixe de clé de tri en utilisant begins_with.

Avant d’écrire une nouvelle entité, définissez ses modèles de clés à l’avance dans une docstring. Le OrderModel dans la section suivante suit cette 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
    """

Stocker plusieurs types d’entités

Le DiscriminatorAttribute de PynamoDB stocke une étiquette de type (entity_type) dans chaque élément. Lors de la requête via BaseModel, cette étiquette est utilisée pour instancier automatiquement chaque résultat en tant que sous-classe correcte — ainsi une seule requête peut retourner un mélange de UserModel, OrderModel et tout autre type d’entité enregistré dans la même table.

Le DiscriminatorAttribute de PynamoDB a été conçu pour le polymorphisme de hiérarchie de classes au sein d’une seule table homogène — pas pour stocker des types d’entités structurellement différents dans une seule table. Le modèle utilisé ici fonctionne, mais gardez ces mises en garde à l’esprit :

La collision de clés est de votre responsabilité. PynamoDB n’empêche pas deux types d’entités d’écrire la même paire pk/sk. Utilisez toujours des préfixes de type uniques (par exemple ORDER#, USER#).
Les requêtes inter-entités nécessitent BaseModel. PynamoDB ajoute automatiquement un filtre de discriminateur lors de la requête d’une sous-classe spécifique, donc les éléments d’autres types d’entités à la même clé ne seront jamais retournés. Pour récupérer un mélange de types à partir d’une clé de partition GSI partagée, interrogez toujours via BaseModel.
Paramètres de table partagés. Toutes les entités partagent la configuration Meta (nom de table, région, identifiants) définie dans BaseModel.

Ci-dessous se trouve un exemple complet à deux entités — un UserModel avec des enregistrements OrderModel associés stockés dans la même table :

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)

Exportez les nouvelles entités depuis __init__.py :

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

Surcharge de GSI

BaseModel fournit deux GSI partagés (gsi1_index, gsi2_index). UserModel et OrderModel ci-dessus écrivent tous deux dans gsi2 — mais avec des valeurs gsi2pk différentes (USER vs ORDER). C’est la surcharge de GSI : réutiliser un seul index physique pour servir plusieurs modèles d’accès indépendants sans consommer de capacité GSI supplémentaire.

UserModel — gsi2pk=USER, gsi2sk=<created_at> → lister tous les utilisateurs par date
OrderModel — gsi2pk=ORDER, gsi2sk=<created_at> → lister toutes les commandes par date

gsi1 peut également être surchargé lorsque plusieurs types d’entités partagent le même parent. Si vous ajoutez plus tard un ReviewModel qui appartient également à un utilisateur, vous pouvez lui attribuer gsi1pk=USER#<user_id> avec une clé de tri REVIEW#<id> — aucun GSI supplémentaire n’est nécessaire. Interroger gsi1 via BaseModel retourne alors à la fois les commandes et les avis pour cet utilisateur en une seule requête, avec PynamoDB instanciant chaque élément en tant que sous-classe correcte :

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

Pour récupérer un seul type d’entité à partir d’un GSI surchargé, utilisez une condition de préfixe de clé de tri :

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

Relations un-à-plusieurs

Dans une relation un-à-plusieurs, l’entité enfant stocke une référence à son parent dans une clé de partition GSI, rendant la relation traversable dans les deux directions sans dupliquer les données. L’exemple UserModel / OrderModel ci-dessus est exactement ce modèle :

Obtenir une seule commande par ID — table primaire : pk=ORDER#<id>, sk=ORDER#<id>
Lister toutes les commandes pour un utilisateur — gsi1 : pk=USER#<user_id>

Une alternative aux recherches basées sur GSI est le modèle de collection d’éléments : donnez aux éléments enfants le même pk que leur parent et utilisez la clé de tri pour les différencier. Cela vous permet de récupérer le parent et tous ses enfants en une seule requête de table primaire, sans GSI :

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

# Retrieve the user and all their orders in one primary-table query
# BaseModel dispatches each item to its correct subclass via 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)]

Le compromis : les collections d’éléments placent tous les enfants sous une seule clé de partition, ce qui est optimal pour la plupart des charges de travail mais peut créer une partition chaude à un débit d’écriture extrême. L’approche GSI (utilisée dans les exemples ci-dessus) garde chaque entité dans sa propre partition et est généralement plus sûre pour commencer.

Relations plusieurs-à-plusieurs

Les relations plusieurs-à-plusieurs nécessitent une entité de jonction utilisant le modèle de liste d’adjacence : un élément dédié qui enregistre chaque lien, avec sa clé GSI inversant la direction afin que la relation puisse être traversée dans les deux sens.

Considérez ArticleModel et TagModel, où un article peut avoir plusieurs tags et un tag peut s’appliquer à plusieurs articles :

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

Parce que ArticleTagModel utilise pk=ARTICLE#<article_id> — la même partition que l’article lui-même — vous pouvez récupérer un article et tous ses tags en une seule requête de table primaire :

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

Pour plus d’informations sur la modélisation des données DynamoDB, consultez le guide de modélisation des données DynamoDB et Création d’une conception à table unique avec Amazon DynamoDB.

Utiliser le client DynamoDB

Le client.py généré exporte deux utilitaires clés :

is_local() — retourne True lorsque LOCAL_DEV=true, utilisé pour basculer entre le comportement local et AWS.
get_table_name() — retourne le nom de la table DynamoDB. Lorsque LOCAL_DEV=true, lit le nom de la table depuis localDev.tableName dans config.json ; sinon récupère le nom depuis AWS AppConfig en utilisant la variable d’environnement RUNTIME_CONFIG_APP_ID et le met en cache pour les appels suivants.

BaseModel dans entities/base.py utilise les deux pour configurer PynamoDB automatiquement :

Connexion — BaseModel.Meta définit host depuis config.json et code en dur region, aws_access_key_id et aws_secret_access_key lorsque is_local() est True, pointant PynamoDB vers l’instance DynamoDB locale. Dans AWS, ceux-ci ne sont pas définis, donc PynamoDB utilise la chaîne d’identifiants par défaut.
Nom de table — BaseModel._get_connection() appelle get_table_name() avant chaque opération, de sorte que la table correcte soit résolue au moment de l’exécution sans aucune configuration manuelle.

Arrêter DynamoDB local

L’arrêt de dev (par exemple avec Ctrl+C) supprime automatiquement le conteneur DynamoDB Local, mais préserve le volume nommé afin que vos données persistent entre les redémarrages.

En raison de limitations dans la gestion des signaux sous Windows, le conteneur n’est pas automatiquement supprimé lorsque dev est arrêté. Vous devrez le supprimer manuellement :

<engine> rm -f <scope>-dynamodb

Remplacez <engine> par votre moteur de conteneurs (docker ou finch) et <scope> par la portée de votre espace de travail Nx (par exemple proj).

Ajouter/Supprimer des index secondaires globaux

Les GSI sont définis dans config.json à la racine du projet sous la clé tableConfig.globalSecondaryIndexes. Ajoutez une entrée pour chaque GSI, puis reflétez le changement dans BaseModel en ajoutant ou supprimant la classe GlobalSecondaryIndex correspondante et les attributs dans <name>/entities/base.py :

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

Le champ sortKey est optionnel pour les GSI avec clé de hachage uniquement.

Ce fichier de configuration est la source unique de vérité lue par tous les consommateurs :

Développement local — dev lit config.json et crée ou met à jour la table locale pour correspondre à la liste des GSI
CDK — le construct lit config.json au moment de la synthèse, donc les modifications de GSI sont reflétées lors du prochain cdk deploy
Terraform — le module lit config.json au moment du plan/apply

Un GSI par Déploiement

Se connecter à la table

Dans n’importe quel projet Python, ajoutez le package DynamoDB en tant que dépendance d’espace de travail et importez directement les classes d’entités :

from my_db_package.entities import ExampleModel

item = ExampleModel.get_by_id('123')

En coulisses, ExampleModel appelle get_table_name() pour récupérer le nom de la table depuis AWS AppConfig au moment de l’exécution.

Déployer votre table

Le générateur DynamoDB crée une infrastructure CDK ou Terraform en fonction de votre iac sélectionné.

CDK
Terraform

Le construct CDK est créé dans common/constructs. Exemple d’utilisation :

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

Cela provisionne une table DynamoDB avec :

pk (clé de partition) et sk (clé de tri), toutes deux de type String
Des index secondaires globaux tels que définis dans config.json
Une facturation à la demande (PAY_PER_REQUEST)
Un chiffrement KMS géré par le client avec rotation automatique des clés
La récupération à un instant dans le passé activée
La protection contre la suppression activée
Le nom de la table enregistré dans Runtime Config sous l’espace de noms dynamodb dans AWS AppConfig

Le module Terraform est créé dans common/terraform. Exemple d’utilisation :

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

Cela provisionne une table DynamoDB avec :

pk (clé de partition) et sk (clé de tri), toutes deux de type String
Des index secondaires globaux tels que définis dans config.json
Une facturation à la demande (PAY_PER_REQUEST)
Un chiffrement KMS géré par le client avec rotation automatique des clés
La récupération à un instant dans le passé activée
La protection contre la suppression activée, plus une garde de cycle de vie prevent_destroy
Le nom de la table enregistré dans Runtime Config sous l’espace de noms dynamodb dans AWS AppConfig

Le module core/runtime-config/appconfig expose l’espace de noms dynamodb par défaut, de sorte que le nom de la table est déployé sans configuration supplémentaire. Si vous passez explicitement namespaces à ce module, conservez dynamodb dans la liste — sinon aucun profil de configuration n’est créé pour celui-ci et le client de table généré ne peut pas résoudre le nom de la table.

Protection contre la suppression

La table est protégée par deux gardes indépendantes, de sorte que la désactivation de l’une seule ne peut pas supprimer vos données :

CDK
Terraform

deletionProtection, appliquée par DynamoDB.
RemovalPolicy.RETAIN, appliquée par CloudFormation, qui laisse la table en place lorsqu’elle est retirée de la stack.

deletion_protection_enabled, appliquée par DynamoDB.
lifecycle { prevent_destroy = true } sur la table dans common/terraform/src/core/dynamodb/dynamodb.tf, appliquée par Terraform, qui fait échouer tout plan qui détruirait la table.

Supprimer la table

Désactivez la protection pour les environnements où la suppression de table est attendue, comme les stacks de développement ou de prévisualisation éphémères.

CDK
Terraform

import { RemovalPolicy } from 'aws-cdk-lib';
import { MyTable } from '@my-scope/common-constructs';

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

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

prevent_destroy doit être un littéral — Terraform ne permet pas de référencer une variable — il ne peut donc pas être désactivé depuis main.tf. Supprimez également le bloc lifecycle de la table dans common/terraform/src/core/dynamodb/dynamodb.tf :

resource "aws_dynamodb_table" "table" {
  # ...

  lifecycle {
    prevent_destroy = true
  }
}

Mode de facturation

La table utilise par défaut la facturation à la demande (PAY_PER_REQUEST). Passez à la capacité provisionnée pour des charges de travail prévisibles à haut débit.

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

Récupération à un instant dans le passé

La récupération à un instant dans le passé est activée par défaut, vous permettant de restaurer la table à n’importe quel moment des 35 derniers jours.

Désactiver la récupération à un instant dans le passé

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
}

Chiffrement

La table est chiffrée avec une clé KMS gérée par le client par défaut, créée automatiquement pour vous. Passez à une clé gérée par AWS, à la clé détenue par AWS, ou apportez votre propre clé KMS, si vous gérez le chiffrement différemment.

Utiliser une clé gérée par AWS

Utilise la clé KMS partagée aws/dynamodb qu’AWS gère en votre nom. Elle est visible dans la console KMS de votre compte et facturée par requête, mais il n’y a pas de clé à créer, faire pivoter ou supprimer.

CDK
Terraform

import { TableEncryption } from 'aws-cdk-lib/aws-dynamodb';
import { MyTable } from '@my-scope/common-constructs';

const table = new MyTable(this, 'Table', {
  encryption: TableEncryption.AWS_MANAGED,
});

Choisir AWS_MANAGED signifie que la table n’est plus chiffrée avec une clé gérée par le client, ce qui échoue à la règle CKV_AWS_119 de Checkov. Supprimez-la sur la table :

import { suppressRules } from '@my-scope/common-constructs';

suppressRules(table.table, ['CKV_AWS_119'], 'Using an AWS managed key rather than a customer-managed CMK');

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

Utiliser la clé détenue par AWS

Utilise une clé entièrement détenue et gérée par AWS — gratuite, sans clé visible dans votre compte. L’option la plus simple lorsque vous n’avez pas besoin d’une clé visible par le client ou le compte pour des raisons de conformité.

CDK
Terraform

import { TableEncryption } from 'aws-cdk-lib/aws-dynamodb';
import { MyTable } from '@my-scope/common-constructs';

const table = new MyTable(this, 'Table', {
  encryption: TableEncryption.DEFAULT,
});

Choisir DEFAULT signifie que la table n’est plus chiffrée avec une clé gérée par le client, ce qui échoue à la règle CKV_AWS_119 de Checkov. Supprimez-la sur la table :

import { suppressRules } from '@my-scope/common-constructs';

suppressRules(table.table, ['CKV_AWS_119'], 'Using the AWS owned key rather than a customer-managed CMK');

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

Passer de CUSTOMER_MANAGED à une autre option

Sur une table déjà déployée, changer encryption de CUSTOMER_MANAGED (vers AWS_MANAGED ou DEFAULT) en un seul terraform apply échoue : Terraform détruit la clé gérée par le client avant de mettre à jour la table, et DynamoDB rejette alors la mise à jour car la clé est déjà en attente de suppression.

Contournez ce problème en mettant à jour le chiffrement de la table directement via l’AWS CLI d’abord, puis en laissant Terraform rattraper son retard et nettoyer la clé orpheline :

# For AWS_MANAGED:
aws dynamodb update-table --table-name <table-name> \
  --sse-specification Enabled=true,SSEType=KMS,KMSMasterKeyId=alias/aws/dynamodb

# For DEFAULT:
aws dynamodb update-table --table-name <table-name> --sse-specification Enabled=false

# Then wait for this to report ENABLED (or for SSEDescription to disappear, for DEFAULT):
aws dynamodb describe-table --table-name <table-name> --query Table.SSEDescription.Status

Ensuite, mettez à jour encryption dans votre configuration Terraform et exécutez terraform apply normalement — Terraform n’a maintenant besoin que de détruire la clé déjà inutilisée, sans rien qui en dépende.

Utiliser votre propre clé KMS

Fournissez une clé gérée par le client existante au lieu d’en faire créer une pour vous. La clé doit déjà accorder au service DynamoDB les permissions dont il a besoin dans sa propre politique de clé.

CDK
Terraform

import { Key } from 'aws-cdk-lib/aws-kms';
import { MyTable } from '@my-scope/common-constructs';

const key = Key.fromKeyArn(this, 'Key', 'arn:aws:kms:us-east-1:111111111111:key/my-key-id');

const table = new MyTable(this, 'Table', {
  encryptionKey: key,
});

module "my_table" {
  source      = "../../common/terraform/src/app/dynamodb/my-table"
  kms_key_arn = "arn:aws:kms:us-east-1:111111111111:key/my-key-id"
}

Rotation de la clé de chiffrement

Lorsque la table crée sa propre clé KMS gérée par le client (par défaut, et uniquement lorsque vous n’avez pas fourni votre propre clé), cette clé a la rotation automatique activée par défaut. Désactivez-la si votre politique de sécurité gère la rotation en externe.

Désactiver la rotation de la clé de chiffrement

CDK
Terraform

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

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

Désactiver la rotation de clé, ou apporter votre propre clé qui ne pivote pas, échoue à la règle CKV_AWS_7 de Checkov. Supprimez-la sur la clé :

import { suppressRules } from '@my-scope/common-constructs';

suppressRules(table.table.encryptionKey!, ['CKV_AWS_7'], 'Key rotation is managed externally');

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

Connexions

Utilisez le générateur connection pour intégrer ce projet avec d’autres dans votre espace de travail. Les connexions suivantes impliquent ce projet :

FastAPI to Python DynamoDBConnect a FastAPI to a DynamoDB table

Python Agent to Python DynamoDBConnect a Python Agent to a DynamoDB table

Python MCP Server to Python DynamoDBConnect a Python MCP Server to a DynamoDB table