Salta ai contenuti

Python DynamoDB

Questo generatore crea un nuovo progetto Python supportato da Amazon DynamoDB, utilizzando PynamoDB per la modellazione delle entità. Genera il codice dell’applicazione e l’infrastruttura necessaria per il provisioning e la gestione di una tabella DynamoDB utilizzando AWS CDK o Terraform, con supporto per il design a tabella singola e sviluppo locale integrato tramite DynamoDB Local.

Terminal window
pnpm nx g @aws/nx-plugin:py#dynamodb
Puoi anche eseguire una prova per vedere quali file verrebbero modificati
Terminal window
pnpm nx g @aws/nx-plugin:py#dynamodb --dry-run
ParametroTipoPredefinitoDescrizione
name Obbligatoriostring-Nome del progetto DynamoDB da generare
directory stringpackagesLa directory in cui memorizzare il progetto.
subDirectory string-La sotto-directory in cui viene posizionato il progetto. Per impostazione predefinita corrisponde al nome del progetto.
framework pynamodbpynamodbIl framework da utilizzare per le entità DynamoDB.
tableName string-Il nome della tabella DynamoDB. Generato automaticamente se non specificato.
infra dynamodb | nonedynamodbInfrastruttura da fornire per la tabella DynamoDB.
iac inherit | cdk | terraforminheritIl provider IaC preferito. Per impostazione predefinita viene ereditato dalla selezione iniziale.
preferInstallDependencies booleantrueSe preferire l'installazione delle dipendenze dopo l'esecuzione del generatore. Impostare su false per rimandare l'installazione quando si eseguono più generatori in batch (l'installazione viene comunque eseguita se necessaria affinché i generatori successivi possano calcolare il grafo dei progetti Nx); installare una volta alla fine.

Il generatore crea la seguente struttura di progetto nella directory <directory>/<name>:

  • Directory<name>
    • __init__.py Package exports
    • client.py DynamoDB client and table name resolution
    • Directoryentities
      • base.py Base PynamoDB model with GSI declarations
      • example.py Example entity definition
      • __init__.py Entity exports
  • config.json Table configuration including GSI definitions and local development settings
  • project.json Project configuration and build targets

Gli script di sviluppo locale sono condivisi tra tutti i progetti DynamoDB (sia TypeScript che Python) e generati una volta in:

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

Poiché questo generatore fornisce infrastruttura come codice basata sul tuo iac scelto, creerà un progetto in packages/common che include i costrutti CDK o i moduli Terraform pertinenti.

Il progetto comune di infrastruttura come codice è strutturato come segue:

  • Directorypackages/common/constructs
    • Directorysrc
      • Directoryapp/ Constructs for infrastructure specific to a project/generator
      • Directorycore/ Generic constructs which are reused by constructs in app
      • index.ts Entry point exporting constructs from app
    • project.json Project build targets and configuration
  • Directorypackages/common/constructs/src
    • Directoryapp
      • Directorydynamodb
        • <name>.ts Infrastruttura specifica per la tua tabella
    • Directorycore
      • dynamodb.ts Costrutto generico per tabelle DynamoDB

Il generatore configura un target dev che avvia un’istanza di DynamoDB Local e crea la tabella. Usa il target dev del progetto:

Terminal window
pnpm nx dev <project-name>

Questo automaticamente:

  1. Scarica l’immagine DynamoDB Local (target pull-image)
  2. Avvia un container
  3. Crea una tabella locale con gli indici definiti in config.json

Il progetto generato utilizza PynamoDB per la modellazione delle entità. Tutte le entità devono ereditare dal BaseModel generato — risolve il nome corretto della tabella DynamoDB a runtime, leggendo da AWS AppConfig quando distribuito o da config.json quando eseguito localmente tramite DynamoDB Local. Senza questo, PynamoDB non saprà quale tabella utilizzare. BaseModel utilizza anche il supporto al polimorfismo di PynamoDB per memorizzare più tipi di entità in una singola tabella, seguendo il design a tabella singola di DynamoDB.

Aggiungi o aggiorna i file delle entità in <name>/entities/, utilizzando l’entità di esempio generata come punto di partenza:

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

Per maggiori dettagli, consulta il tutorial di PynamoDB.

In DynamoDB, la progettazione dello schema inizia con le tue query, non con la forma dei tuoi dati. Prima di scrivere qualsiasi modello, elenca ogni pattern di accesso di cui la tua applicazione ha bisogno, quindi progetta i valori delle chiavi pk, sk e GSI in modo che ogni pattern sia risposto da una singola richiesta alla tabella — nessun JOIN, nessuna lettura sequenziale.

L’ExampleModel generato dimostra questo per tre pattern:

  • Get by ID — indice primario, pk=EXAMPLE#<id>, sk=EXAMPLE#<id>
  • List by categorygsi1, pk=CATEGORY#<category>
  • List by creation dategsi2, pk=EXAMPLE, chiave di ordinamento tra timestamp ISO

La convenzione del prefisso di tipo (ad es. EXAMPLE#, CATEGORY#) è deliberata: rende gli elementi auto-descrittivi quando si naviga nella tabella, previene collisioni accidentali di chiavi tra tipi di entità che condividono un indice e consente il filtraggio dei prefissi delle chiavi di ordinamento utilizzando begins_with.

Prima di scrivere una nuova entità, definisci i suoi pattern di chiavi in anticipo in una docstring. L’OrderModel nella sezione successiva segue questa convenzione:

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

Il DiscriminatorAttribute di PynamoDB memorizza un’etichetta di tipo (entity_type) in ogni elemento. Quando si esegue una query tramite BaseModel, questa etichetta viene utilizzata per istanziare automaticamente ogni risultato come la sua sottoclasse corretta — quindi una singola query può restituire un mix di UserModel, OrderModel e qualsiasi altro tipo di entità registrato nella stessa tabella.

Di seguito è riportato un esempio completo con due entità — un UserModel con record OrderModel associati memorizzati nella stessa tabella:

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)

Esporta le nuove entità da __init__.py:

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

BaseModel fornisce due GSI condivisi (gsi1_index, gsi2_index). Sia UserModel che OrderModel sopra scrivono su gsi2 — ma con valori gsi2pk diversi (USER vs ORDER). Questo è overloading dei GSI: riutilizzare un singolo indice fisico per servire più pattern di accesso indipendenti senza consumare capacità GSI extra.

  • UserModelgsi2pk=USER, gsi2sk=<created_at> → elenca tutti gli utenti per data
  • OrderModelgsi2pk=ORDER, gsi2sk=<created_at> → elenca tutti gli ordini per data

gsi1 può anche essere sovraccaricato quando più tipi di entità condividono lo stesso genitore. Se successivamente aggiungi un ReviewModel che appartiene anche a un utente, puoi assegnargli gsi1pk=USER#<user_id> con una chiave di ordinamento REVIEW#<id> — nessun GSI aggiuntivo necessario. Interrogare gsi1 tramite BaseModel restituisce quindi sia ordini che recensioni per quell’utente in una richiesta, con PynamoDB che istanzia ogni elemento come la sua sottoclasse corretta:

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

Per recuperare solo un tipo di entità da un GSI sovraccaricato, utilizza una condizione di prefisso della chiave di ordinamento:

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

In una relazione uno-a-molti l’entità figlia memorizza un riferimento al suo genitore in una chiave di partizione GSI, rendendo la relazione attraversabile in entrambe le direzioni senza duplicare i dati. L’esempio UserModel / OrderModel sopra è esattamente questo pattern:

  • Ottieni un singolo ordine per ID — tabella primaria: pk=ORDER#<id>, sk=ORDER#<id>
  • Elenca tutti gli ordini per un utentegsi1: pk=USER#<user_id>

Un’alternativa alle ricerche basate su GSI è il pattern item collection: dai agli elementi figli lo stesso pk del loro genitore e usa la chiave di ordinamento per differenziarli. Questo ti consente di recuperare il genitore e tutti i suoi figli in una singola query sulla tabella primaria, senza un 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)]

Il compromesso: le item collection posizionano tutti i figli sotto una singola chiave di partizione, che è ottimale per la maggior parte dei carichi di lavoro ma può creare una partizione calda con throughput di scrittura estremo. L’approccio GSI (utilizzato negli esempi sopra) mantiene ogni entità nella propria partizione ed è generalmente più sicuro per iniziare.

Le relazioni molti-a-molti richiedono un’entità di giunzione utilizzando il pattern adjacency list: un elemento dedicato che registra ogni collegamento, con la sua chiave GSI che inverte la direzione in modo che la relazione possa essere attraversata in entrambi i modi.

Considera ArticleModel e TagModel, dove un articolo può avere molti tag e un tag può applicarsi a molti articoli:

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

Poiché ArticleTagModel utilizza pk=ARTICLE#<article_id> — la stessa partizione dell’articolo stesso — puoi recuperare un articolo e tutti i suoi tag in una singola query sulla tabella 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)]

Per ulteriori letture sulla modellazione dei dati DynamoDB, consulta la guida alla modellazione dei dati DynamoDB e Creating a single-table design with Amazon DynamoDB.

Il client.py generato esporta due utility chiave:

  • is_local() — restituisce True quando LOCAL_DEV=true, utilizzato per passare tra comportamento locale e AWS.
  • get_table_name() — restituisce il nome della tabella DynamoDB. Quando LOCAL_DEV=true, legge il nome della tabella da localDev.tableName in config.json; altrimenti recupera il nome da AWS AppConfig utilizzando la variabile d’ambiente RUNTIME_CONFIG_APP_ID e lo memorizza nella cache per le chiamate successive.

BaseModel in entities/base.py utilizza entrambi per configurare PynamoDB automaticamente:

  • ConnessioneBaseModel.Meta imposta host da config.json e codifica region, aws_access_key_id e aws_secret_access_key quando is_local() è True, puntando PynamoDB all’istanza DynamoDB locale. In AWS, questi vengono lasciati non impostati in modo che PynamoDB utilizzi la catena di credenziali predefinita.
  • Nome tabellaBaseModel._get_connection() chiama get_table_name() prima di ogni operazione, quindi la tabella corretta viene risolta a runtime senza alcuna configurazione manuale.

L’arresto di dev (ad esempio con Ctrl+C) rimuove automaticamente il container DynamoDB Local, ma preserva il volume nominato in modo che i tuoi dati persistano tra i riavvii.

I GSI sono definiti in config.json nella radice del progetto sotto la chiave tableConfig.globalSecondaryIndexes. Aggiungi una voce per ogni GSI, quindi rifletti la modifica in BaseModel aggiungendo o rimuovendo la classe GlobalSecondaryIndex corrispondente e gli attributi in <name>/entities/base.py:

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

Il campo sortKey è facoltativo per i GSI con solo hash-key.

Questo file di configurazione è l’unica fonte di verità letta da tutti i consumatori:

  • Sviluppo localedev legge config.json e crea o aggiorna la tabella locale per corrispondere all’elenco dei GSI
  • CDK — il costrutto legge config.json al momento della sintesi, quindi le modifiche ai GSI si riflettono al prossimo cdk deploy
  • Terraform — il modulo legge config.json al momento del plan/apply

In qualsiasi progetto Python, aggiungi il pacchetto DynamoDB come dipendenza del workspace e importa direttamente le classi delle entità:

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

Dietro le quinte, ExampleModel chiama get_table_name() per recuperare il nome della tabella da AWS AppConfig a runtime.

Il generatore DynamoDB crea infrastruttura CDK o Terraform in base al tuo iac selezionato.

Il costrutto CDK viene creato in common/constructs. Esempio di utilizzo:

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

Questo esegue il provisioning di una tabella DynamoDB con:

  • pk (partition key) e sk (sort key), entrambe di tipo String
  • Global Secondary Indexes come definiti in config.json
  • Fatturazione on-demand (PAY_PER_REQUEST)
  • Crittografia KMS gestita dal cliente con rotazione automatica delle chiavi
  • Point-in-time recovery abilitato
  • Protezione dalla cancellazione abilitata
  • Nome della tabella registrato in Runtime Config sotto il namespace dynamodb in AWS AppConfig

La protezione dalla cancellazione è abilitata per impostazione predefinita per prevenire la cancellazione accidentale della tabella.

Disabilitala per ambienti in cui è prevista la cancellazione della tabella, come stack di sviluppo o preview di breve durata.

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

La tabella utilizza per impostazione predefinita la fatturazione on-demand (PAY_PER_REQUEST). Passa alla capacità con provisioning per carichi di lavoro prevedibili ad alto throughput.

packages/infra/src/stacks/application-stack.ts
import { BillingMode } from 'aws-cdk-lib/aws-dynamodb';
import { MyTable } from '@my-scope/common-constructs';
const table = new MyTable(this, 'Table', {
billingMode: BillingMode.PROVISIONED,
readCapacity: 5,
writeCapacity: 5,
});

Il Point-in-time recovery è abilitato per impostazione predefinita, consentendoti di ripristinare la tabella a qualsiasi momento negli ultimi 35 giorni.

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

La chiave KMS utilizzata per crittografare la tabella ha la rotazione automatica delle chiavi abilitata per impostazione predefinita. Disabilitala se la tua policy di sicurezza gestisce la rotazione esternamente.

Disabilitare la rotazione delle chiavi di crittografia

Sezione intitolata “Disabilitare la rotazione delle chiavi di crittografia”
packages/infra/src/stacks/application-stack.ts
import { MyTable } from '@my-scope/common-constructs';
const table = new MyTable(this, 'Table', {
enableKeyRotation: false,
});

Utilizza il generatore connection per integrare questo progetto con altri nel tuo workspace. Le seguenti connessioni coinvolgono questo progetto:

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