py#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

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

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

Bạn cũng có thể thực hiện chạy thử để xem những tệp nào sẽ bị thay đổi

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

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

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

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

Tùy chọn

Tham số	Kiểu	Mặc định	Mô tả
name Bắt buộc	string	-	Tên của dự án DynamoDB cần tạo
directory	string	packages	Thư mục để lưu trữ dự án.
subDirectory	string	-	Thư mục con nơi dự án được đặt. Mặc định là tên dự án.
framework	pynamodb	pynamodb	Framework sử dụng cho các entity DynamoDB.
tableName	string	-	Tên bảng DynamoDB. Tự động tạo nếu không chỉ định.
infra	dynamodb \| none	dynamodb	Hạ tầng cần cung cấp cho bảng DynamoDB.
iac	inherit \| cdk \| terraform	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.
preferInstallDependencies	boolean	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ả 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ẻ trên 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 iacProvider 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 cho infrastructure cụ thể của một dự án/generator
    …
  - Thư mụccore/ Constructs chung được tái sử dụng bởi các constructs trong app
    …
  - index.ts Entry point xuất các constructs từ app
- project.json Các build targets và cấu hình của dự án

Thư mụcpackages/common/terraform
- Thư mụcsrc
  - Thư mụcapp/ Terraform modules cho infrastructure cụ thể của một dự án/generator
    …
  - Thư mụccore/ Modules chung được tái sử dụng bởi các modules trong app
    …
- project.json Các build targets và cấu hình của dự án

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

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ó phân giải 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ế Dựa trên Mẫu Truy cập

Trong DynamoDB, thiết kế schema bắt đầu từ 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ỳ model 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 timestamp 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 định nghĩa 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 thông 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 — do đó 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ể, do đó 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 thông 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 định nghĩa trong BaseModel.

Dưới đây là một ví dụ đầy đủ 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 thông 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 quá tải, 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: 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, 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
    """
    ...

# Truy xuất người dùng và tất cả đơn hàng của họ trong một truy vấn bảng chính
# BaseModel phân phối mỗi mục đến lớp con chính xác của nó thông qua 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

Tệp 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, các giá trị này không được đặt để 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, do đó bảng chính xác được phân giải tại thời điểm chạy mà không cần cấu hình thủ công.

Dừng DynamoDB Cục bộ

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 tự động bị xóa khi dev dừng lại. 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 định nghĩa 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 và các thuộc tính tương ứng 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 đối với các GSI chỉ có hash-key.

Tệp 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:

Local development — 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 sẽ đượ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 cho mỗi Deployment

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à import trực tiếp các lớp thực thể:

from my_db_package.entities import ExampleModel

item = ExampleModel.get_by_id('123')

Connection Generators

Đối với các loại dự án cụ thể, sử dụng generator connection để tự động kết nối các phụ thuộc phát triển cục bộ để DynamoDB Local khởi động tự động cùng với dự án của bạn, và thêm gói DynamoDB làm phụ thuộc workspace:

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

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

CDK
Terraform

CDK construct đượ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 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 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

Cấp quyền truy cập

Để cho phép các hàm Lambda truy cập bảng DynamoDB, hãy cấp các quyền cần thiết trong cơ sở hạ tầng của bạn.

CDK
Terraform

Gọi grantReadWriteData trên table construct. Điều này cấp cả quyền DynamoDB và KMS cần thiết cho vai trò thực thi Lambda:

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

const table = new MyTable(this, 'Table');

const api = new Api(this, 'Api', {
  integrations: Api.defaultIntegrations(this).build(),
});

Object.entries(api.integrations).forEach(([, integration]) => {
  table.grantReadWriteData(integration.handler);
});

Cấp quyền cho vai trò thực thi Lambda để truy cập bảng DynamoDB và khóa mã hóa KMS của nó:

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

resource "aws_iam_role_policy" "dynamodb_access" {
  role = module.my_api.lambda_role_name

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "dynamodb:GetItem",
          "dynamodb:PutItem",
          "dynamodb:UpdateItem",
          "dynamodb:DeleteItem",
          "dynamodb:Query",
          "dynamodb:Scan",
          "dynamodb:BatchGetItem",
          "dynamodb:BatchWriteItem",
        ]
        Resource = [
          module.my_table.table_arn,
          "${module.my_table.table_arn}/index/*",
        ]
      },
      {
        Effect = "Allow"
        Action = [
          "kms:Encrypt",
          "kms:Decrypt",
          "kms:ReEncrypt*",
          "kms:GenerateDataKey*",
          "kms:DescribeKey"
        ]
        Resource = [module.my_table.kms_key_arn]
      },
    ]
  })
}

Bảo vệ xóa

Bảo vệ xóa được bật mặc định để ngăn chặn việc xóa bảng do nhầm lẫn.

Tắt bảo vệ xóa

Tắt nó 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 tồn tại ngắn hạn.

CDK
Terraform

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

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

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

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.

Tắt khôi phục theo thời điểm

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
}

Xoay khóa mã hóa

Khóa KMS được sử dụng để mã hóa bảng có tự động xoay khóa được bật mặc định. Tắt nó nếu chính sách bảo mật của bạn quản lý việc xoay khóa từ bên ngoài.

Tắt xoay khóa mã hóa

CDK
Terraform

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

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

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

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 DynamoDB Kết nối FastAPI với bảng DynamoDB

Python Agent to Python DynamoDB Kết nối Python Agent với bảng DynamoDB

Python MCP Server to Python DynamoDB Kết nối Python MCP Server với bảng DynamoDB