AWS DynamoDB Python CRUD

验证：未记录
适用版本
未记录
验证范围
尚未登记验证记录。

本文缩写（英文全称与中文含义）
缩写 英文全称／名称 中文含义
API Application Programming Interface 应用程序编程接口
AWS Amazon Web Services 亚马逊云服务
CRUD Create, Read, Update, Delete 创建、读取、更新、删除
GSI Global Secondary Index DynamoDB 全局二级索引
HKD Hong Kong Dollar 港元货币代码
IAM Identity and Access Management 身份与访问管理
S3 Simple Storage Service AWS 对象存储服务
SDK Software Development Kit 软件开发工具包
ULID Universally Unique Lexicographically Sortable Identifier 可按字典顺序排序的全局唯一标识符
UUID Universally Unique Identifier 通用唯一标识符

1. Important Points#

这个 page 用一个真实但简化的场景演示 DynamoDB CRUD：订单服务里的「用户订单状态表」。

业务场景：

service:
    order-api

feature:
    用户提交订单后，App 需要快速查询订单状态
    支付、仓储、配送系统会更新订单状态和 updated_at
    客服后台按 customer_id 查看用户订单

main access patterns:
    1. create order by customer_id + order_id
    2. get one order by customer_id + order_id
    3. list orders by customer_id
    4. update order updated_at / status by customer_id + order_id
    5. delete test / expired order in non-production

为什么这个场景可以用 DynamoDB：

fit:
    query pattern 很固定，主要按 customer_id 和 order_id 访问
    不需要 join 用户表、商品表、支付表才能返回订单状态
    单条订单 item 小于 400KB
    读写量可能很高，需要低延迟 key-value / document access
    每次请求可以直接命中 partition key，不需要 ad-hoc query

not the full order system:
    payment ledger should not be only this table
    financial reconciliation usually needs relational / ledger model
    analytics should export to S3 / warehouse, not Scan online table

在线请求不要依赖全表 Scan 完成分析查询；按访问模式设计 key/index，离线分析走 S3 或数仓。

什么时候不要用 DynamoDB 做这个功能：

prefer MySQL / PostgreSQL when:
    需要复杂 join: user + order + order_items + invoice
    需要 ad-hoc filter / sort: by city, coupon, amount, channel, staff
    需要复杂报表和聚合
    事务跨很多实体并且强一致语义复杂

prefer DynamoDB when:
    API 只需要 get/list/update by known key
    查询维度少且稳定
    要承受高并发读写
    可以接受通过 GSI / stream / export 支撑其他读模型

这里的 customer_order_status 不是完整电商订单数据库，而是 order-api 的在线状态读写表。生产设计重点不是 boto3 语法，而是 key design。

table:
    name: customer_order_status
    partition key: customer_id
    sort key: order_id

example:
    customer_id = cus_8f3a91
    order_id = ord_20260618_0001
    status = PAID
    total_amount = 129.90
    currency = HKD
    shipping_city = Hong Kong

why these keys:
    customer_id groups one customer's orders into one item collection
    order_id identifies one order under that customer
    GetItem can fetch one order directly with customer_id + order_id
    Query can list all orders for one customer by customer_id

order_id choice:
    use sortable order_id if the list page needs newest-first order
    example: ord_20260618_0001 or ULID
    if order_id is random UUID, use a sort key like created_at_order_id instead

query:
    always query by partition key
    avoid Scan for online API
    use ConditionExpression to protect create / update / delete

pk / sk 是通用 key 名，常用于 single-table design。本页是单实体示例，用 customer_id / order_id 更直观。

single-table example:
    partition key: pk
    sort key: sk
    pk = CUSTOMER#cus_8f3a91
    sk = ORDER#ord_20260618_0001

why use this:
    one table stores multiple entity types such as CUSTOMER / ORDER / PAYMENT
    prefix makes entity type visible in the key
    begins_with(sk, "ORDER#") can select order items under one customer

for this CRUD page:
    use meaningful key names first
    do not introduce pk/sk unless the article is about single-table design

2. IAM#

Application role 最小权限示例：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "dynamodb:PutItem",
        "dynamodb:GetItem",
        "dynamodb:Query",
        "dynamodb:UpdateItem",
        "dynamodb:DeleteItem"
      ],
      "Resource": "arn:aws:dynamodb:ap-east-1:111122223333:table/customer_order_status"
    }
  ]
}

3. Dependency#

python3 -m venv .venv
source .venv/bin/activate
pip install boto3

Runtime config:

export AWS_REGION=ap-northeast-1
export TABLE_NAME=customer_order_status

4. Create Table#

aws dynamodb create-table --region "${AWS_REGION}" --table-name "${TABLE_NAME}" --attribute-definitions AttributeName=customer_id,AttributeType=S AttributeName=order_id,AttributeType=S --key-schema AttributeName=customer_id,KeyType=HASH AttributeName=order_id,KeyType=RANGE --billing-mode PAY_PER_REQUEST

5. Code#

dynamodb_crud.py:

import os
from datetime import datetime, timezone
from decimal import Decimal

import boto3
from botocore.exceptions import ClientError


REGION = os.getenv("AWS_REGION", "ap-northeast-1")
TABLE_NAME = os.getenv("TABLE_NAME", "customer_order_status")

client = boto3.client("dynamodb", region_name=REGION)
dynamodb = boto3.resource("dynamodb", region_name=REGION)
table = dynamodb.Table(TABLE_NAME)


def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def ensure_table() -> None:
    try:
        client.describe_table(TableName=TABLE_NAME)
        return
    except ClientError as e:
        if e.response["Error"]["Code"] != "ResourceNotFoundException":
            raise

    client.create_table(
        TableName=TABLE_NAME,
        AttributeDefinitions=[
            {"AttributeName": "customer_id", "AttributeType": "S"},
            {"AttributeName": "order_id", "AttributeType": "S"},
        ],
        KeySchema=[
            {"AttributeName": "customer_id", "KeyType": "HASH"},
            {"AttributeName": "order_id", "KeyType": "RANGE"},
        ],
        BillingMode="PAY_PER_REQUEST",
    )

    waiter = client.get_waiter("table_exists")
    waiter.wait(TableName=TABLE_NAME)


def put_order(customer_id: str, order_id: str, total_amount: str) -> None:
    ts = now_iso()
    item = {
        "customer_id": customer_id,
        "order_id": order_id,
        "status": "PAID",
        "total_amount": Decimal(total_amount),
        "currency": "HKD",
        "payment_id": "pay_20260618_0001",
        "warehouse_id": "hk-east-1",
        "shipping_city": "Hong Kong",
        "tracking_number": None,
        "created_at": ts,
        "updated_at": ts,
    }

    table.put_item(
        Item=item,
        ConditionExpression="attribute_not_exists(customer_id) AND attribute_not_exists(order_id)",
    )


def get_order_eventual(customer_id: str, order_id: str) -> dict | None:
    resp = table.get_item(
        Key={
            "customer_id": customer_id,
            "order_id": order_id,
        }
    )
    return resp.get("Item")


def get_order_strong(customer_id: str, order_id: str) -> dict | None:
    resp = table.get_item(
        Key={
            "customer_id": customer_id,
            "order_id": order_id,
        },
        ConsistentRead=True,
    )
    return resp.get("Item")


def transact_get_order_and_payment(customer_id: str, order_id: str, payment_id: str) -> list[dict]:
    resp = client.transact_get_items(
        TransactItems=[
            {
                "Get": {
                    "TableName": TABLE_NAME,
                    "Key": {
                        "customer_id": {"S": customer_id},
                        "order_id": {"S": order_id},
                    },
                }
            },
            {
                "Get": {
                    "TableName": TABLE_NAME,
                    "Key": {
                        "customer_id": {"S": customer_id},
                        "order_id": {"S": f"PAYMENT#{payment_id}"},
                    },
                }
            },
        ]
    )
    return [item["Item"] for item in resp["Responses"] if "Item" in item]


def list_customer_orders(customer_id: str) -> list[dict]:
    resp = table.query(
        KeyConditionExpression="customer_id = :customer_id",
        ExpressionAttributeValues={
            ":customer_id": customer_id,
        },
        ScanIndexForward=False,
    )
    return resp["Items"]


def update_order_status(customer_id: str, order_id: str, status: str) -> dict:
    resp = table.update_item(
        Key={
            "customer_id": customer_id,
            "order_id": order_id,
        },
        UpdateExpression="SET #status = :status, updated_at = :updated_at",
        ConditionExpression="attribute_exists(customer_id) AND attribute_exists(order_id)",
        ExpressionAttributeNames={
            "#status": "status",
        },
        ExpressionAttributeValues={
            ":status": status,
            ":updated_at": now_iso(),
        },
        ReturnValues="ALL_NEW",
    )
    return resp["Attributes"]


def transact_create_order_and_payment(customer_id: str, order_id: str, payment_id: str) -> None:
    ts = now_iso()
    client.transact_write_items(
        TransactItems=[
            {
                "Put": {
                    "TableName": TABLE_NAME,
                    "Item": {
                        "customer_id": {"S": customer_id},
                        "order_id": {"S": order_id},
                        "entity_type": {"S": "ORDER"},
                        "status": {"S": "PAID"},
                        "total_amount": {"N": "129.90"},
                        "currency": {"S": "HKD"},
                        "payment_id": {"S": payment_id},
                        "created_at": {"S": ts},
                        "updated_at": {"S": ts},
                    },
                    "ConditionExpression": "attribute_not_exists(customer_id) AND attribute_not_exists(order_id)",
                }
            },
            {
                "Put": {
                    "TableName": TABLE_NAME,
                    "Item": {
                        "customer_id": {"S": customer_id},
                        "order_id": {"S": f"PAYMENT#{payment_id}"},
                        "entity_type": {"S": "PAYMENT"},
                        "payment_id": {"S": payment_id},
                        "status": {"S": "CAPTURED"},
                        "created_at": {"S": ts},
                    },
                    "ConditionExpression": "attribute_not_exists(customer_id) AND attribute_not_exists(order_id)",
                }
            },
        ]
    )


def delete_order(customer_id: str, order_id: str) -> None:
    table.delete_item(
        Key={
            "customer_id": customer_id,
            "order_id": order_id,
        },
        ConditionExpression="attribute_exists(customer_id) AND attribute_exists(order_id)",
    )


if __name__ == "__main__":
    try:
        ensure_table()
        put_order("cus_8f3a91", "ord_20260618_0001", "129.90")
        print("eventual read:", get_order_eventual("cus_8f3a91", "ord_20260618_0001"))
        print("strong read:", get_order_strong("cus_8f3a91", "ord_20260618_0001"))
        print()
        print("query:", list_customer_orders("cus_8f3a91"))
        print()
        print("update:", update_order_status("cus_8f3a91", "ord_20260618_0001", "SHIPPED"))
        print()
        transact_create_order_and_payment("cus_8f3a91", "ord_20260618_0002", "pay_20260618_0002")
        print("transactional read:", transact_get_order_and_payment("cus_8f3a91", "ord_20260618_0002", "pay_20260618_0002"))
        delete_order("cus_8f3a91", "ord_20260618_0001")
        delete_order("cus_8f3a91", "ord_20260618_0002")
        delete_order("cus_8f3a91", "PAYMENT#pay_20260618_0002")
        print("deleted")
    except ClientError as e:
        print(e.response["Error"]["Code"], e.response["Error"]["Message"])
        raise

这个示例里几种读写方式的对应关系：

eventually consistent read:
    get_order_eventual() uses GetItem without ConsistentRead

strongly consistent read:
    get_order_strong() uses GetItem with ConsistentRead=True

transactional read:
    transact_get_order_and_payment() uses TransactGetItems

standard write:
    put_order() / update_order_status() use PutItem / UpdateItem

transactional write:
    transact_create_order_and_payment() uses TransactWriteItems

Run:

-> % python dynamodb_crud.py
eventual read: {... 'status': 'PAID', 'customer_id': 'cus_8f3a91', 'order_id': 'ord_20260618_0001'}
strong read: {... 'status': 'PAID', 'customer_id': 'cus_8f3a91', 'order_id': 'ord_20260618_0001'}

query: [{... 'order_id': 'ord_20260618_0001'}]

update: {... 'status': 'SHIPPED', 'order_id': 'ord_20260618_0001'}

transactional read: [
  {'customer_id': {'S': 'cus_8f3a91'}, 'order_id': {'S': 'ord_20260618_0002'}, ...},
  {'customer_id': {'S': 'cus_8f3a91'}, 'order_id': {'S': 'PAYMENT#pay_20260618_0002'}, ...}
]
deleted

本地运行这段代码通常不会稳定复现 stale read；get_order_eventual() / get_order_strong() 的重点是展示 API 参数差异。

6. Production Notes#

use Decimal:
    DynamoDB number type should not use Python float

use conditional writes:
    create: attribute_not_exists
    update/delete: attribute_exists
    idempotency: condition on request_id / version

read consistency:
    GetItem can use ConsistentRead=True when needed
    Query on GSI is eventually consistent only

lookup by order_id only:
    add a GSI with order_id as partition key
    do not Scan the table from online API

do not:
    Scan from online request path
    store unbounded item collections under one hot partition key
    retry ConditionalCheckFailedException blindly
Export S3 Forward

搜索文档

输入关键词，搜索所有文档。

↑ ↓ 选择 · Enter 打开 · Esc 关闭