Python CRUD


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

什么时候不要用 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