Python DynamoDB

이 생성기는 Amazon DynamoDB를 기반으로 하는 새로운 Python 프로젝트를 생성하며, 엔티티 모델링을 위해 PynamoDB를 사용합니다. AWS CDK 또는 Terraform을 사용하여 DynamoDB 테이블을 프로비저닝하고 관리하는 데 필요한 애플리케이션 코드와 인프라를 생성하며, 단일 테이블 설계 지원과 DynamoDB Local을 통한 로컬 개발 기능이 내장되어 있습니다.

사용법

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

어떤 파일이 변경될지 확인하기 위해 드라이 런을 수행할 수도 있습니다

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

설치 Nx Console VSCode Plugin 아직 설치하지 않았다면
VSCode에서 Nx 콘솔 열기
클릭 Generate (UI) "Common Nx Commands" 섹션에서
검색 @aws/nx-plugin - py#dynamodb
필수 매개변수 입력
클릭 Generate

옵션

매개변수	타입	기본값	설명
name 필수	string	-	생성할 DynamoDB 프로젝트의 이름
directory	string	packages	프로젝트를 저장할 디렉토리입니다.
subDirectory	string	-	프로젝트가 배치되는 하위 디렉토리입니다. 기본값은 프로젝트 이름입니다.
framework	pynamodb	pynamodb	DynamoDB 엔티티에 사용할 프레임워크입니다.
tableName	string	-	DynamoDB 테이블 이름입니다. 지정하지 않으면 자동 생성됩니다.
infra	dynamodb \| none	dynamodb	DynamoDB 테이블을 위해 프로비저닝할 인프라입니다.
iac	inherit \| cdk \| terraform	inherit	선호하는 IaC 공급자입니다. 기본값은 초기 선택에서 상속됩니다.
preferInstallDependencies	boolean	true	생성기 실행 후 의존성 설치를 선호할지 여부입니다. 여러 생성기를 일괄 처리할 때 설치를 연기하려면 false로 설정하세요(후속 생성기가 Nx 프로젝트 그래프를 계산할 수 있도록 필요한 경우 설치는 여전히 실행됩니다). 마지막에 한 번만 설치하세요.

생성기 출력

생성기는 <directory>/<name> 디렉토리에 다음과 같은 프로젝트 구조를 생성합니다:

디렉터리<name>
- __init__.py Package exports
- client.py DynamoDB client and table name resolution
- 디렉터리entities
  - 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

로컬 개발 스크립트는 모든 DynamoDB 프로젝트(TypeScript 및 Python 모두)에서 공유되며 다음 위치에 한 번 생성됩니다:

디렉터리packages/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

인프라

이 생성기는 선택한 iac를 기반으로 코드형 인프라를 제공하므로, 관련 CDK constructs 또는 Terraform 모듈을 포함하는 packages/common에 프로젝트를 생성합니다.

공통 코드형 인프라 프로젝트는 다음과 같이 구성됩니다:

CDK
Terraform

디렉터리packages/common/constructs
- 디렉터리src
  - 디렉터리app/ 프로젝트/생성기에 특정한 인프라를 위한 Constructs
    …
  - 디렉터리core/ app의 constructs에서 재사용되는 일반 constructs
    …
  - index.ts app에서 constructs를 내보내는 진입점
- project.json 프로젝트 빌드 타겟 및 구성

디렉터리packages/common/terraform
- 디렉터리src
  - 디렉터리app/ 프로젝트/생성기에 특정한 인프라를 위한 Terraform 모듈
    …
  - 디렉터리core/ app의 모듈에서 재사용되는 일반 모듈
    …
- project.json 프로젝트 빌드 타겟 및 구성

CDK
Terraform

디렉터리packages/common/constructs/src
- 디렉터리app
  - 디렉터리dynamodb
    <name>.ts 테이블에 특정한 인프라
- 디렉터리core
  - dynamodb.ts 범용 DynamoDB 테이블 구성

로컬 개발

로컬 DynamoDB 시작

생성기는 DynamoDB Local 인스턴스를 시작하고 테이블을 생성하는 dev 타겟을 구성합니다. 프로젝트의 dev 타겟을 사용하세요:

pnpm nx dev <project-name>

yarn nx dev <project-name>

npx nx dev <project-name>

bunx nx dev <project-name>

이것은 자동으로:

DynamoDB Local 이미지를 가져옵니다 (pull-image 타겟)
컨테이너를 시작합니다
config.json에 정의된 인덱스로 로컬 테이블을 생성합니다

데이터 모델링

생성된 프로젝트는 엔티티 모델링을 위해 PynamoDB를 사용합니다. 모든 엔티티는 생성된 BaseModel을 반드시 상속해야 합니다. BaseModel은 런타임에 올바른 DynamoDB 테이블 이름을 확인하며, 배포 시에는 AWS AppConfig에서 읽고 DynamoDB Local을 통해 로컬에서 실행할 때는 config.json에서 읽습니다. 이것이 없으면 PynamoDB는 어떤 테이블을 사용해야 하는지 알 수 없습니다. BaseModel은 또한 PynamoDB의 다형성 지원을 사용하여 DynamoDB의 단일 테이블 설계를 따라 단일 테이블에 여러 엔티티 유형을 저장합니다.

생성된 예제 엔티티를 시작점으로 사용하여 <name>/entities/ 아래에 엔티티 파일을 추가하거나 업데이트하세요:

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

자세한 내용은 PynamoDB 튜토리얼을 참조하세요.

액세스 패턴을 중심으로 설계하기

DynamoDB에서 스키마 설계는 데이터 형태가 아닌 쿼리에서 시작됩니다. 모델을 작성하기 전에 애플리케이션에 필요한 모든 액세스 패턴을 나열한 다음, 각 패턴이 단일 테이블 요청으로 응답될 수 있도록 pk, sk 및 GSI 키 값을 설계하세요. JOIN이나 순차 읽기는 필요하지 않습니다.

생성된 ExampleModel은 세 가지 패턴에 대해 이를 보여줍니다:

ID로 가져오기 — 기본 인덱스, pk=EXAMPLE#<id>, sk=EXAMPLE#<id>
카테고리별 목록 — gsi1, pk=CATEGORY#<category>
생성 날짜별 목록 — gsi2, pk=EXAMPLE, ISO 타임스탬프 사이의 정렬 키

타입 접두사 규칙(예: EXAMPLE#, CATEGORY#)은 의도적입니다. 테이블을 탐색할 때 항목을 자체 설명적으로 만들고, 인덱스를 공유하는 엔티티 유형 간의 우발적인 키 충돌을 방지하며, begins_with를 사용한 정렬 키 접두사 필터링을 허용합니다.

새 엔티티를 작성하기 전에 docstring에 키 패턴을 미리 정의하세요. 다음 섹션의 OrderModel은 이 규칙을 따릅니다:

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

여러 엔티티 유형 저장

PynamoDB의 DiscriminatorAttribute는 모든 항목에 타입 레이블(entity_type)을 저장합니다. BaseModel을 통해 쿼리할 때 이 레이블은 각 결과를 올바른 서브클래스로 자동으로 인스턴스화하는 데 사용됩니다. 따라서 단일 쿼리가 동일한 테이블에 등록된 UserModel, OrderModel 및 기타 엔티티 유형의 혼합을 반환할 수 있습니다.

PynamoDB의 DiscriminatorAttribute는 단일 동질 테이블 내의 클래스 계층 다형성을 위해 설계되었으며, 하나의 테이블에 구조적으로 다른 엔티티 유형을 저장하기 위한 것이 아닙니다. 여기서 사용된 패턴은 작동하지만 다음 주의사항을 염두에 두세요:

키 충돌은 사용자의 책임입니다. PynamoDB는 두 엔티티 유형이 동일한 pk/sk 쌍을 작성하는 것을 방지하지 않습니다. 항상 고유한 타입 접두사(예: ORDER#, USER#)를 사용하세요.
교차 엔티티 쿼리에는 BaseModel이 필요합니다. PynamoDB는 특정 서브클래스를 쿼리할 때 자동으로 판별자 필터를 추가하므로 동일한 키에 있는 다른 엔티티 유형의 항목은 절대 반환되지 않습니다. 공유 GSI 파티션 키에서 여러 유형의 혼합을 검색하려면 항상 BaseModel을 통해 쿼리하세요.
공유 테이블 설정. 모든 엔티티는 BaseModel에 정의된 Meta 구성(테이블 이름, 리전, 자격 증명)을 공유합니다.

다음은 동일한 테이블에 저장된 연관된 OrderModel 레코드가 있는 UserModel의 완전한 두 엔티티 예제입니다:

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)

__init__.py에서 새 엔티티를 내보내세요:

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

GSI 오버로딩

BaseModel은 두 개의 공유 GSI(gsi1_index, gsi2_index)를 제공합니다. 위의 UserModel과 OrderModel 모두 gsi2에 쓰지만 서로 다른 gsi2pk 값(USER 대 ORDER)을 사용합니다. 이것이 GSI 오버로딩입니다: 추가 GSI 용량을 소비하지 않고 여러 독립적인 액세스 패턴을 제공하기 위해 단일 물리적 인덱스를 재사용하는 것입니다.

UserModel — gsi2pk=USER, gsi2sk=<created_at> → 날짜별로 모든 사용자 나열
OrderModel — gsi2pk=ORDER, gsi2sk=<created_at> → 날짜별로 모든 주문 나열

여러 엔티티 유형이 동일한 부모를 공유할 때 gsi1도 오버로드될 수 있습니다. 나중에 사용자에게도 속하는 ReviewModel을 추가하는 경우 REVIEW#<id> 정렬 키와 함께 gsi1pk=USER#<user_id>를 할당할 수 있습니다. 추가 GSI가 필요하지 않습니다. BaseModel을 통해 gsi1을 쿼리하면 해당 사용자의 주문과 리뷰를 모두 한 번의 요청으로 반환하며, PynamoDB는 각 항목을 올바른 서브클래스로 인스턴스화합니다:

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

오버로드된 GSI에서 하나의 엔티티 유형만 검색하려면 정렬 키 접두사 조건을 사용하세요:

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

일대다 관계

일대다 관계에서 자식 엔티티는 GSI 파티션 키에 부모에 대한 참조를 저장하여 데이터를 복제하지 않고 양방향으로 관계를 탐색할 수 있게 합니다. 위의 UserModel / OrderModel 예제가 정확히 이 패턴입니다:

ID로 단일 주문 가져오기 — 기본 테이블: pk=ORDER#<id>, sk=ORDER#<id>
사용자의 모든 주문 나열 — gsi1: pk=USER#<user_id>

GSI 기반 조회의 대안은 항목 컬렉션 패턴입니다: 자식 항목에 부모와 동일한 pk를 부여하고 정렬 키를 사용하여 구분합니다. 이렇게 하면 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)]

트레이드오프: 항목 컬렉션은 모든 자식을 단일 파티션 키 아래에 배치하므로 대부분의 워크로드에 최적이지만 극단적인 쓰기 처리량에서 핫 파티션을 생성할 수 있습니다. GSI 접근 방식(위의 예제에서 사용됨)은 각 엔티티를 자체 파티션에 유지하며 일반적으로 시작하기에 더 안전합니다.

다대다 관계

다대다 관계에는 인접 목록 패턴을 사용하는 연결 엔티티가 필요합니다: 각 링크를 기록하는 전용 항목으로, GSI 키가 방향을 반전시켜 관계를 양방향으로 탐색할 수 있습니다.

기사가 여러 태그를 가질 수 있고 태그가 여러 기사에 적용될 수 있는 ArticleModel과 TagModel을 고려하세요:

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

ArticleTagModel은 pk=ARTICLE#<article_id>를 사용하므로(기사 자체와 동일한 파티션) 단일 기본 테이블 쿼리로 기사와 모든 태그를 검색할 수 있습니다:

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

DynamoDB 데이터 모델링에 대한 추가 정보는 DynamoDB 데이터 모델링 가이드 및 Amazon DynamoDB로 단일 테이블 설계 만들기를 참조하세요.

DynamoDB 클라이언트 사용

생성된 client.py는 두 가지 주요 유틸리티를 내보냅니다:

is_local() — LOCAL_DEV=true일 때 True를 반환하며, 로컬과 AWS 동작 간 전환에 사용됩니다.
get_table_name() — DynamoDB 테이블 이름을 반환합니다. LOCAL_DEV=true일 때는 config.json의 localDev.tableName에서 테이블 이름을 읽고, 그렇지 않으면 RUNTIME_CONFIG_APP_ID 환경 변수를 사용하여 AWS AppConfig에서 이름을 가져와 후속 호출을 위해 캐시합니다.

entities/base.py의 BaseModel은 두 가지를 모두 사용하여 PynamoDB를 자동으로 구성합니다:

연결 — BaseModel.Meta는 is_local()이 True일 때 config.json에서 host를 설정하고 region, aws_access_key_id, aws_secret_access_key를 하드코딩하여 PynamoDB를 로컬 DynamoDB 인스턴스로 지정합니다. AWS에서는 이러한 값이 설정되지 않으므로 PynamoDB는 기본 자격 증명 체인을 사용합니다.
테이블 이름 — BaseModel._get_connection()은 각 작업 전에 get_table_name()을 호출하므로 수동 구성 없이 런타임에 올바른 테이블이 확인됩니다.

로컬 DynamoDB 중지

dev를 중지하면 (예: Ctrl+C로) DynamoDB Local 컨테이너가 자동으로 제거되지만, 명명된 볼륨은 보존되므로 재시작 시에도 데이터가 유지됩니다.

Windows의 신호 처리 제한으로 인해 dev가 중지될 때 컨테이너가 자동으로 제거되지 않습니다. 수동으로 제거해야 합니다:

<engine> rm -f <scope>-dynamodb

<engine>을 컨테이너 엔진(docker 또는 finch)으로, <scope>를 Nx 워크스페이스 스코프(예: proj)로 바꾸세요.

글로벌 보조 인덱스 추가/제거

GSI는 tableConfig.globalSecondaryIndexes 키 아래 프로젝트 루트의 config.json에 정의됩니다. 각 GSI에 대한 항목을 추가한 다음 <name>/entities/base.py에서 해당 GlobalSecondaryIndex 클래스와 속성을 추가하거나 제거하여 BaseModel에 변경 사항을 반영하세요:

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

sortKey 필드는 해시 키만 사용하는 GSI의 경우 선택 사항입니다.

이 구성 파일은 모든 소비자가 읽는 단일 진실 공급원입니다:

로컬 개발 — dev는 config.json을 읽고 GSI 목록과 일치하도록 로컬 테이블을 생성하거나 업데이트합니다
CDK — 구성체는 합성 시점에 config.json을 읽으므로 GSI 변경 사항이 다음 cdk deploy에 반영됩니다
Terraform — 모듈은 plan/apply 시점에 config.json을 읽습니다

배포당 하나의 GSI

테이블에 연결

모든 Python 프로젝트에서 DynamoDB 패키지를 작업 공간 종속성으로 추가하고 엔티티 클래스를 직접 가져오세요:

from my_db_package.entities import ExampleModel

item = ExampleModel.get_by_id('123')

내부적으로 ExampleModel은 get_table_name()을 호출하여 런타임에 AWS AppConfig에서 테이블 이름을 가져옵니다.

테이블 배포

DynamoDB 생성기는 선택한 iac에 따라 CDK 또는 Terraform 인프라를 생성합니다.

CDK
Terraform

CDK 구성은 common/constructs에 생성됩니다. 사용 예시:

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

다음과 같은 DynamoDB 테이블을 프로비저닝합니다:

pk (파티션 키)와 sk (정렬 키), 둘 다 String 타입
config.json에 정의된 글로벌 보조 인덱스
온디맨드 (PAY_PER_REQUEST) 요금제
자동 키 로테이션이 활성화된 고객 관리형 KMS 암호화
특정 시점 복구 활성화
삭제 방지 활성화
AWS AppConfig의 dynamodb 네임스페이스 아래 Runtime Config에 등록된 테이블 이름

Terraform 모듈은 common/terraform에 생성됩니다. 사용 예시:

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

다음과 같은 DynamoDB 테이블을 프로비저닝합니다:

pk (파티션 키)와 sk (정렬 키), 둘 다 String 타입
config.json에 정의된 글로벌 보조 인덱스
온디맨드 (PAY_PER_REQUEST) 요금제
자동 키 로테이션이 활성화된 고객 관리형 KMS 암호화
특정 시점 복구 활성화
삭제 방지 활성화
Runtime Config에 등록된 테이블 이름

삭제 방지

실수로 테이블이 삭제되는 것을 방지하기 위해 삭제 방지가 기본적으로 활성화되어 있습니다.

삭제 방지 비활성화

단기 개발 또는 프리뷰 스택과 같이 테이블 삭제가 예상되는 환경에서는 비활성화하세요.

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
}

요금제 모드

테이블은 기본적으로 온디맨드 (PAY_PER_REQUEST) 요금제를 사용합니다. 예측 가능한 고처리량 워크로드의 경우 프로비저닝된 용량으로 전환하세요.

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

특정 시점 복구

특정 시점 복구는 기본적으로 활성화되어 있으며, 지난 35일 동안의 임의 시점으로 테이블을 복원할 수 있습니다.

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
}

암호화 키 로테이션

테이블을 암호화하는 데 사용되는 KMS 키는 기본적으로 자동 키 로테이션이 활성화되어 있습니다. 보안 정책에서 로테이션을 외부적으로 관리하는 경우 비활성화하세요.

암호화 키 로테이션 비활성화

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
}

연결

connection 생성기를 사용하여 이 프로젝트를 작업 공간의 다른 프로젝트와 통합하세요. 다음 연결에는 이 프로젝트가 포함됩니다:

FastAPI to Python DynamoDBFastAPI를 DynamoDB 테이블에 연결

Python Agent to Python DynamoDBPython Agent를 DynamoDB 테이블에 연결

Python MCP Server to Python DynamoDBPython MCP Server를 DynamoDB 테이블에 연결