Aller au contenu

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.

Terminal window
pnpm nx g @aws/nx-plugin:py#dynamodb
Vous pouvez également effectuer une simulation pour voir quels fichiers seraient modifiés
Terminal window
pnpm nx g @aws/nx-plugin:py#dynamodb --dry-run
ParamètreTypePar défautDescription
name Requisstring-Nom du projet DynamoDB à générer
directory stringpackagesLe répertoire dans lequel stocker le projet.
subDirectory string-Le sous-répertoire dans lequel le projet est placé. Par défaut, il s'agit du nom du projet.
framework pynamodbpynamodbLe framework à utiliser pour les entités DynamoDB.
tableName string-Le nom de la table DynamoDB. Généré automatiquement s'il n'est pas spécifié.
infra dynamodb | nonedynamodbInfrastructure à provisionner pour la table DynamoDB.
iac inherit | cdk | terraforminheritLe fournisseur IaC préféré. Par défaut, il est hérité de votre sélection initiale.
preferInstallDependencies booleantrueIndique 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.

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

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

  • 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/constructs/src
    • Répertoireapp
      • Répertoiredynamodb
        • <name>.ts Infrastructure spécifique à votre table
    • Répertoirecore
      • dynamodb.ts Construct générique de table DynamoDB

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 :

Terminal window
pnpm nx dev <project-name>

Cela effectue automatiquement :

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

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 :

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

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

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égoriegsi1, pk=CATEGORY#<category>
  • Lister par date de créationgsi2, 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
"""

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.

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 :

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)

Exportez les nouvelles entités depuis __init__.py :

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

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.

  • UserModelgsi2pk=USER, gsi2sk=<created_at> → lister tous les utilisateurs par date
  • OrderModelgsi2pk=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#'),
))

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

packages/my_table/my_table/entities/order.py (item collection variant)
class OrderModel(BaseModel, discriminator='OrderModel'):
"""
Key design (item collection):
pk=USER#<user_id>, sk=ORDER#<order_id> <- co-located under the parent user
"""
...
# Retrieve the user and all their orders in one primary-table query
# BaseModel dispatches each item to its correct subclass via DiscriminatorAttribute
items = list(BaseModel.query(f'USER#{user_id}'))
user = next(i for i in items if isinstance(i, UserModel))
orders = [i for i in items if isinstance(i, OrderModel)]

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.

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 :

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

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.

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 :

  • ConnexionBaseModel.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 tableBaseModel._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.

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.

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 :

config.json
{
...
"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 localdev 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

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.

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

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

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

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 donné 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

La protection contre la suppression est activée par défaut pour empêcher la suppression accidentelle de la table.

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

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

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

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

La récupération à un instant donné 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 donné

Section intitulée « Désactiver la récupération à un instant donné »
packages/infra/src/stacks/application-stack.ts
import { MyTable } from '@my-scope/common-constructs';
const table = new MyTable(this, 'Table', {
pointInTimeRecoverySpecification: { pointInTimeRecoveryEnabled: false },
});

La clé KMS utilisée pour chiffrer la table a la rotation automatique des clés activée par défaut. Désactivez-la si votre politique de sécurité gère la rotation en externe.

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

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 :

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