Python DynamoDB

Generator này tạo một dự án Python mới được hỗ trợ bởi Amazon DynamoDB, sử dụng PynamoDB để mô hình hóa thực thể. Nó tạo ra mã ứng dụng và cơ sở hạ tầng cần thiết để cung cấp và quản lý bảng DynamoDB bằng AWS CDK hoặc Terraform, với hỗ trợ thiết kế bảng đơn và phát triển cục bộ tích hợp sẵn thông qua DynamoDB Local.

Cách sử dụng

Tạo một dự án DynamoDB

Chạy generator này@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

Cài đặt Nx Console VSCode Plugin nếu bạn chưa cài đặt
Mở Nx Console trong VSCode
Nhấp Generate (UI) trong phần "Common Nx Commands"
Tìm kiếm @aws/nx-plugin - py#dynamodb
Điền các tham số bắt buộc
Nhấp Generate

Xây dựng lệnh của bạn8

nameBắt buộc

Tùy chọn

Tùy chọn của generator8 tùy chọn

nameBắt buộcstring: Tên của dự án DynamoDB cần tạo
directorystringMặc định: packages: Thư mục để lưu trữ dự án.
frameworkenumMặc định: pynamodb: Framework sử dụng cho các entity DynamoDB.
pynamodb
infraenumMặc định: dynamodb: Hạ tầng cần cung cấp cho bảng DynamoDB.
dynamodbnone
iacenumMặc định: inherit: Nhà cung cấp IaC ưu tiên. Mặc định được kế thừa từ lựa chọn ban đầu của bạn.
inheritcdkterraform
subDirectorystring: Thư mục con nơi dự án được đặt. Mặc định là tên dự án.
tableNamestring: Tên bảng DynamoDB. Tự động tạo nếu không chỉ định.
preferInstallDependenciesbooleanMặc định: true: Có nên cài đặt các phụ thuộc sau khi generator chạy hay không. Đặt thành false để hoãn việc cài đặt khi chạy nhiều generator liên tiếp (việc cài đặt vẫn sẽ chạy nếu cần thiết để các generator tiếp theo có thể tính toán đồ thị dự án Nx); cài đặt một lần vào cuối.

Kết quả của Generator

Generator tạo cấu trúc dự án sau trong thư mục <directory>/<name>:

Thư mục<name>
- __init__.py Package exports
- client.py DynamoDB client and table name resolution
- Thư mụcentities
  - 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

Các script phát triển cục bộ được chia sẻ giữa tất cả các dự án DynamoDB (cả TypeScript và Python) và được tạo một lần vào:

Thư mụcpackages/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

Cơ sở hạ tầng

Vì generator này cung cấp infrastructure as code dựa trên iac bạn đã chọn, nó sẽ tạo một dự án trong packages/common bao gồm các CDK constructs hoặc Terraform modules liên quan.

Dự án infrastructure as code chung được cấu trúc như sau:

CDK
Terraform

Thư mụcpackages/common/constructs
- Thư mụcsrc
  - Thư mụcapp/ Constructs for infrastructure specific to a project/generator
    …
  - Thư mụccore/ Generic constructs which are reused by constructs in app
    …
  - index.ts Entry point exporting constructs from app
- project.json Project build targets and configuration

Thư mụcpackages/common/terraform
- Thư mụcsrc
  - Thư mụcapp/ Terraform modules for infrastructure specific to a project/generator
    …
  - Thư mụccore/ Generic modules which are reused by modules in app
    …
- project.json Project build targets and configuration

CDK
Terraform

Thư mụcpackages/common/constructs/src
- Thư mụcapp
  - Thư mụcdynamodb
    <name>.ts Cơ sở hạ tầng cụ thể cho bảng của bạn
- Thư mụccore
  - dynamodb.ts Construct bảng DynamoDB chung

Kiến trúc

Dự án được triển khai cung cấp chính bảng đó, mà bất kỳ dự án nào được kết nối với nó đều đọc và ghi:

Loading the diagram…

Phát triển cục bộ

Khởi động DynamoDB cục bộ

Generator cấu hình một target dev để khởi động một instance DynamoDB Local và tạo bảng. Sử dụng target dev của dự án:

pnpm nx dev <project-name>

yarn nx dev <project-name>

npx nx dev <project-name>

bunx nx dev <project-name>

Điều này tự động:

Kéo image DynamoDB Local (target pull-image)
Khởi động một container
Tạo một bảng local với các index được định nghĩa trong config.json

Mô hình hóa dữ liệu

Dự án được tạo sử dụng PynamoDB để mô hình hóa thực thể. Tất cả các thực thể phải kế thừa từ BaseModel được tạo — nó giải quyết tên bảng DynamoDB chính xác tại thời điểm chạy, đọc từ AWS AppConfig khi triển khai hoặc từ config.json khi chạy cục bộ thông qua DynamoDB Local. Nếu không có điều này, PynamoDB sẽ không biết bảng nào để sử dụng. BaseModel cũng sử dụng hỗ trợ đa hình của PynamoDB để lưu trữ nhiều loại thực thể trong một bảng duy nhất, tuân theo thiết kế bảng đơn của DynamoDB.

Thêm hoặc cập nhật các tệp thực thể trong <name>/entities/, sử dụng thực thể ví dụ được tạo làm điểm khởi đầu:

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

Để biết thêm chi tiết, xem hướng dẫn PynamoDB.

Thiết kế xung quanh các mẫu truy cập

Trong DynamoDB, thiết kế schema bắt đầu với các truy vấn của bạn, không phải hình dạng dữ liệu. Trước khi viết bất kỳ mô hình nào, hãy liệt kê mọi mẫu truy cập mà ứng dụng của bạn cần, sau đó thiết kế các giá trị khóa pk, sk và GSI sao cho mỗi mẫu được trả lời bằng một yêu cầu bảng duy nhất — không có JOIN, không có đọc tuần tự.

ExampleModel được tạo minh họa điều này cho ba mẫu:

Lấy theo ID — chỉ mục chính, pk=EXAMPLE#<id>, sk=EXAMPLE#<id>
Liệt kê theo danh mục — gsi1, pk=CATEGORY#<category>
Liệt kê theo ngày tạo — gsi2, pk=EXAMPLE, khóa sắp xếp giữa các dấu thời gian ISO

Quy ước tiền tố loại (ví dụ: EXAMPLE#, CATEGORY#) là có chủ đích: nó làm cho các mục tự mô tả khi duyệt bảng, ngăn chặn va chạm khóa ngẫu nhiên giữa các loại thực thể chia sẻ một chỉ mục, và cho phép lọc tiền tố khóa sắp xếp bằng cách sử dụng begins_with.

Trước khi viết một thực thể mới, hãy xác định các mẫu khóa của nó trước trong một docstring. OrderModel trong phần tiếp theo tuân theo quy ước này:

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

Lưu trữ nhiều loại thực thể

DiscriminatorAttribute của PynamoDB lưu trữ một nhãn loại (entity_type) trong mọi mục. Khi truy vấn qua BaseModel, nhãn này được sử dụng để khởi tạo mỗi kết quả dưới dạng lớp con chính xác của nó một cách tự động — vì vậy một truy vấn duy nhất có thể trả về hỗn hợp UserModel, OrderModel và bất kỳ loại thực thể nào khác được đăng ký trong cùng một bảng.

DiscriminatorAttribute của PynamoDB được thiết kế cho đa hình phân cấp lớp trong một bảng đồng nhất duy nhất — không phải để lưu trữ các loại thực thể khác nhau về cấu trúc trong một bảng. Mẫu được sử dụng ở đây hoạt động, nhưng hãy ghi nhớ những lưu ý sau:

Va chạm khóa là trách nhiệm của bạn. PynamoDB không ngăn chặn hai loại thực thể ghi cùng một cặp pk/sk. Luôn sử dụng các tiền tố loại duy nhất (ví dụ: ORDER#, USER#).
Truy vấn xuyên thực thể yêu cầu BaseModel. PynamoDB tự động thêm bộ lọc discriminator khi truy vấn một lớp con cụ thể, vì vậy các mục của các loại thực thể khác tại cùng một khóa sẽ không bao giờ được trả về. Để truy xuất hỗn hợp các loại từ một khóa phân vùng GSI được chia sẻ, luôn truy vấn qua BaseModel.
Cài đặt bảng được chia sẻ. Tất cả các thực thể chia sẻ cấu hình Meta (tên bảng, vùng, thông tin xác thực) được xác định trong BaseModel.

Dưới đây là một ví dụ hoàn chỉnh về hai thực thể — một UserModel với các bản ghi OrderModel liên quan được lưu trữ trong cùng một bảng:

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)

Xuất các thực thể mới từ __init__.py:

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

Quá tải GSI

BaseModel cung cấp hai GSI được chia sẻ (gsi1_index, gsi2_index). Cả UserModel và OrderModel ở trên đều ghi vào gsi2 — nhưng với các giá trị gsi2pk khác nhau (USER so với ORDER). Đây là quá tải GSI: tái sử dụng một chỉ mục vật lý duy nhất để phục vụ nhiều mẫu truy cập độc lập mà không tiêu tốn thêm dung lượng GSI.

UserModel — gsi2pk=USER, gsi2sk=<created_at> → liệt kê tất cả người dùng theo ngày
OrderModel — gsi2pk=ORDER, gsi2sk=<created_at> → liệt kê tất cả đơn hàng theo ngày

gsi1 cũng có thể được quá tải khi nhiều loại thực thể chia sẻ cùng một cha. Nếu sau này bạn thêm một ReviewModel cũng thuộc về một người dùng, bạn có thể gán cho nó gsi1pk=USER#<user_id> với khóa sắp xếp REVIEW#<id> — không cần thêm GSI. Truy vấn gsi1 qua BaseModel sau đó trả về cả đơn hàng và đánh giá cho người dùng đó trong một yêu cầu, với PynamoDB khởi tạo mỗi mục dưới dạng lớp con chính xác của nó:

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

Để chỉ truy xuất một loại thực thể từ một GSI được quá tải, hãy sử dụng điều kiện tiền tố khóa sắp xếp:

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

Quan hệ một-nhiều

Trong quan hệ một-nhiều, thực thể con lưu trữ một tham chiếu đến cha của nó trong một khóa phân vùng GSI, làm cho quan hệ có thể duyệt theo cả hai hướng mà không cần sao chép dữ liệu. Ví dụ UserModel / OrderModel ở trên chính xác là mẫu này:

Lấy một đơn hàng duy nhất theo ID — bảng chính: pk=ORDER#<id>, sk=ORDER#<id>
Liệt kê tất cả đơn hàng cho một người dùng — gsi1: pk=USER#<user_id>

Một giải pháp thay thế cho tra cứu dựa trên GSI là mẫu bộ sưu tập mục: cung cấp cho các mục con cùng pk với cha của chúng và sử dụng khóa sắp xếp để phân biệt chúng. Điều này cho phép bạn truy xuất cha và tất cả các con của nó trong một truy vấn bảng chính duy nhất, mà không cần 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)]

Sự đánh đổi: bộ sưu tập mục đặt tất cả các con dưới một khóa phân vùng duy nhất, điều này tối ưu cho hầu hết các khối lượng công việc nhưng có thể tạo ra một phân vùng nóng ở thông lượng ghi cực cao. Cách tiếp cận GSI (được sử dụng trong các ví dụ ở trên) giữ mỗi thực thể trong phân vùng riêng của nó và thường an toàn hơn để bắt đầu.

Quan hệ nhiều-nhiều

Quan hệ nhiều-nhiều yêu cầu một thực thể nối sử dụng mẫu danh sách kề: một mục chuyên dụng ghi lại mỗi liên kết, với khóa GSI của nó đảo ngược hướng để quan hệ có thể được duyệt theo cả hai cách.

Xem xét ArticleModel và TagModel, trong đó một bài viết có thể có nhiều thẻ và một thẻ có thể áp dụng cho nhiều bài viết:

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

Bởi vì ArticleTagModel sử dụng pk=ARTICLE#<article_id> — cùng phân vùng với chính bài viết — bạn có thể truy xuất một bài viết và tất cả các thẻ của nó trong một truy vấn bảng chính duy nhất:

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

Để đọc thêm về mô hình hóa dữ liệu DynamoDB, xem hướng dẫn mô hình hóa dữ liệu DynamoDB và Tạo thiết kế bảng đơn với Amazon DynamoDB.

Sử dụng DynamoDB Client

client.py được tạo xuất hai tiện ích chính:

is_local() — trả về True khi LOCAL_DEV=true, được sử dụng để chuyển đổi giữa hành vi cục bộ và AWS.
get_table_name() — trả về tên bảng DynamoDB. Khi LOCAL_DEV=true, đọc tên bảng từ localDev.tableName trong config.json; nếu không, lấy tên từ AWS AppConfig bằng biến môi trường RUNTIME_CONFIG_APP_ID và lưu vào bộ nhớ cache cho các lần gọi tiếp theo.

BaseModel trong entities/base.py sử dụng cả hai để cấu hình PynamoDB tự động:

Kết nối — BaseModel.Meta đặt host từ config.json và mã hóa cứng region, aws_access_key_id, và aws_secret_access_key khi is_local() là True, trỏ PynamoDB đến phiên bản DynamoDB cục bộ. Trong AWS, những thứ này được để trống để PynamoDB sử dụng chuỗi thông tin xác thực mặc định.
Tên bảng — BaseModel._get_connection() gọi get_table_name() trước mỗi thao tác, vì vậy bảng chính xác được giải quyết tại thời điểm chạy mà không cần bất kỳ cấu hình thủ công nào.

Dừng DynamoDB cục bộ

Việc dừng dev (ví dụ: bằng Ctrl+C) sẽ tự động xóa container DynamoDB Local, nhưng vẫn giữ lại named volume để dữ liệu của bạn được bảo toàn qua các lần khởi động lại.

Do hạn chế trong việc xử lý tín hiệu trên Windows, container sẽ không được tự động xóa khi dev bị dừng. Bạn sẽ cần phải xóa nó thủ công:

<engine> rm -f <scope>-dynamodb

Thay thế <engine> bằng container engine của bạn (docker hoặc finch) và <scope> bằng scope của Nx workspace (ví dụ: proj).

Thêm/Xóa Global Secondary Indexes

GSI được xác định trong config.json tại thư mục gốc dự án dưới khóa tableConfig.globalSecondaryIndexes. Thêm một mục cho mỗi GSI, sau đó phản ánh thay đổi trong BaseModel bằng cách thêm hoặc xóa lớp GlobalSecondaryIndex tương ứng và các thuộc tính trong <name>/entities/base.py:

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

Trường sortKey là tùy chọn cho các GSI chỉ có hash-key.

File cấu hình này là nguồn sự thật duy nhất được đọc bởi tất cả các consumer:

Phát triển cục bộ — dev đọc config.json và tạo hoặc cập nhật bảng cục bộ để khớp với danh sách GSI
CDK — construct đọc config.json tại thời điểm synth, do đó các thay đổi GSI được phản ánh trong lần cdk deploy tiếp theo
Terraform — module đọc config.json tại thời điểm plan/apply

Một GSI mỗi Triển khai

Kết nối đến bảng

Trong bất kỳ dự án Python nào, thêm gói DynamoDB làm phụ thuộc workspace và nhập các lớp thực thể trực tiếp:

from my_db_package.entities import ExampleModel

item = ExampleModel.get_by_id('123')

Đằng sau hậu trường, ExampleModel gọi get_table_name() để lấy tên bảng từ AWS AppConfig tại thời điểm chạy.

Triển khai bảng của bạn

Generator DynamoDB tạo cơ sở hạ tầng CDK hoặc Terraform dựa trên iac bạn đã chọn.

CDK
Terraform

Construct CDK được tạo trong common/constructs. Ví dụ sử dụng:

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

Điều này cung cấp một bảng DynamoDB với:

pk (partition key) và sk (sort key), cả hai đều là kiểu String
Global Secondary Indexes như được định nghĩa trong config.json
Thanh toán theo yêu cầu (PAY_PER_REQUEST)
Mã hóa KMS do khách hàng quản lý với tự động xoay vòng khóa
Khôi phục theo thời điểm được bật
Bảo vệ xóa được bật
Tên bảng được đăng ký trong Runtime Config dưới namespace dynamodb trong AWS AppConfig

Module Terraform được tạo trong common/terraform. Ví dụ sử dụng:

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

Điều này cung cấp một bảng DynamoDB với:

pk (partition key) và sk (sort key), cả hai đều là kiểu String
Global Secondary Indexes như được định nghĩa trong config.json
Thanh toán theo yêu cầu (PAY_PER_REQUEST)
Mã hóa KMS do khách hàng quản lý với tự động xoay vòng khóa
Khôi phục theo thời điểm được bật
Bảo vệ xóa được bật, cùng với bảo vệ vòng đời prevent_destroy
Tên bảng được đăng ký trong Runtime Config dưới namespace dynamodb trong AWS AppConfig

Module core/runtime-config/appconfig hiển thị namespace dynamodb theo mặc định, vì vậy tên bảng được triển khai mà không cần cấu hình thêm. Nếu bạn truyền namespaces vào module đó một cách rõ ràng, hãy giữ dynamodb trong danh sách — nếu không, không có configuration profile nào được tạo cho nó và table client được tạo ra không thể phân giải tên bảng.

Bảo vệ Xóa

Bảng được bảo vệ bởi hai biện pháp bảo vệ độc lập, do đó việc tắt một trong hai không thể xóa dữ liệu của bạn:

CDK
Terraform

deletionProtection, được thực thi bởi DynamoDB.
RemovalPolicy.RETAIN, được thực thi bởi CloudFormation, giữ bảng tại chỗ khi nó bị xóa khỏi stack.

deletion_protection_enabled, được thực thi bởi DynamoDB.
lifecycle { prevent_destroy = true } trên bảng trong common/terraform/src/core/dynamodb/dynamodb.tf, được thực thi bởi Terraform, làm thất bại bất kỳ kế hoạch nào sẽ phá hủy bảng.

Xóa Bảng

Vô hiệu hóa bảo vệ cho các môi trường mà việc xóa bảng được mong đợi, chẳng hạn như các stack phát triển hoặc xem trước có thời gian tồn tại ngắn.

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 phải là một giá trị cố định — Terraform không cho phép nó tham chiếu đến một biến — vì vậy nó không thể được tắt từ main.tf. Cũng xóa khối lifecycle khỏi bảng trong common/terraform/src/core/dynamodb/dynamodb.tf:

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

  lifecycle {
    prevent_destroy = true
  }
}

Chế độ Thanh toán

Bảng mặc định sử dụng thanh toán theo yêu cầu (PAY_PER_REQUEST). Chuyển sang dung lượng được cung cấp cho các khối lượng công việc có thông lượng cao và có thể dự đoán.

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

Khôi phục Theo Thời điểm

Khôi phục theo thời điểm được bật mặc định, cho phép bạn khôi phục bảng về bất kỳ thời điểm nào trong 35 ngày qua.

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
}

Mã hóa

Bảng được mã hóa với khóa KMS do khách hàng quản lý theo mặc định, được tạo tự động cho bạn. Chuyển sang khóa do AWS quản lý, khóa thuộc sở hữu của AWS, hoặc mang khóa KMS của riêng bạn, nếu bạn quản lý mã hóa theo cách khác.

Sử dụng Khóa do AWS Quản lý

Sử dụng khóa KMS aws/dynamodb được chia sẻ mà AWS quản lý thay mặt bạn. Nó hiển thị trong bảng điều khiển KMS của tài khoản bạn và được tính phí theo yêu cầu, nhưng không có khóa nào để bạn tạo, xoay vòng hoặc xóa.

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

Chọn AWS_MANAGED có nghĩa là bảng không còn được mã hóa bằng khóa do khách hàng quản lý, điều này làm thất bại CKV_AWS_119 của Checkov. Loại bỏ nó trên bảng:

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

Sử dụng Khóa Thuộc sở hữu của AWS

Sử dụng một khóa hoàn toàn thuộc sở hữu và được quản lý bởi AWS — miễn phí, không có khóa nào hiển thị trong tài khoản của bạn. Tùy chọn đơn giản nhất khi bạn không cần khóa hiển thị cho khách hàng hoặc tài khoản vì lý do tuân thủ.

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

Chọn DEFAULT có nghĩa là bảng không còn được mã hóa bằng khóa do khách hàng quản lý, điều này làm thất bại CKV_AWS_119 của Checkov. Loại bỏ nó trên bảng:

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

Chuyển đổi khỏi CUSTOMER_MANAGED

Trên bảng đã triển khai, việc thay đổi encryption khỏi CUSTOMER_MANAGED (sang AWS_MANAGED hoặc DEFAULT) trong một lệnh terraform apply duy nhất sẽ thất bại: Terraform phá hủy khóa do khách hàng quản lý trước khi cập nhật bảng, và DynamoDB sau đó từ chối cập nhật vì khóa đã đang chờ xóa.

Giải quyết bằng cách cập nhật mã hóa của bảng trực tiếp qua AWS CLI trước, sau đó để Terraform bắt kịp và dọn dẹp khóa bị bỏ rơi:

# 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

Sau đó cập nhật encryption trong cấu hình Terraform của bạn và chạy terraform apply như bình thường — Terraform bây giờ chỉ cần phá hủy khóa đã không sử dụng, không còn gì phụ thuộc vào nó.

Sử dụng Khóa KMS của Riêng Bạn

Cung cấp một khóa do khách hàng quản lý hiện có thay vì để một khóa được tạo cho bạn. Khóa phải đã cấp cho dịch vụ DynamoDB các quyền cần thiết trong chính sách khóa của nó.

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

Xoay vòng Khóa Mã hóa

Khi bảng tạo khóa KMS do khách hàng quản lý của riêng nó (mặc định, và chỉ khi bạn chưa cung cấp khóa của riêng mình), khóa đó có tính năng xoay vòng khóa tự động được bật theo mặc định. Vô hiệu hóa nó nếu chính sách bảo mật của bạn quản lý xoay vòng từ bên ngoài.

Vô hiệu hóa Xoay vòng Khóa Mã hóa

CDK
Terraform

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

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

Vô hiệu hóa xoay vòng khóa, hoặc mang khóa của riêng bạn không xoay vòng, làm thất bại CKV_AWS_7 của Checkov. Loại bỏ nó trên khóa:

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
}

Kết nối

Sử dụng generator connection để tích hợp dự án này với các dự án khác trong workspace của bạn. Các kết nối sau liên quan đến dự án này:

FastAPI to Python DynamoDBKết nối một FastAPI đến một bảng DynamoDB

Python Agent to Python DynamoDBKết nối một Python Agent đến một bảng DynamoDB

Python MCP Server to Python DynamoDBKết nối một Python MCP Server đến một bảng DynamoDB