pynamodb，一个神奇的 Python 库！

第一次用 boto3 写 DynamoDB，我最烦的不是建表，是写那堆 ExpressionAttributeNames、ExpressionAttributeValues。

明明只是改一个订单状态，代码看起来像在拼一份报文。

UpdateExpression="SET #s = :s, updated_at = :t"
ExpressionAttributeNames={"#s": "status"}
ExpressionAttributeValues={":s": "PAID", ":t": now}

这东西能用，但写多了手疼。更麻烦的是，业务代码里到处散落 DynamoDB 的字段名，哪天字段一改，搜全项目都搜得不放心。

后来我在一个 Python 服务里用了 pynamodb，第一眼感觉：这玩意不是多高级，是它把 DynamoDB 那套啰嗦 API 压住了。官方也把它定位成 Amazon DynamoDB 的 Pythonic interface，PyPI 上的描述也很直接：DynamoDB 很强，但 API 偏啰嗦，PynamoDB 给了一层更简单的接口。

先看一段像业务里真会出现的代码。

import os
from datetime import datetime, timezone
from pynamodb.attributes import UnicodeAttribute, UTCDateTimeAttribute, NumberAttribute
from pynamodb.indexes import GlobalSecondaryIndex, AllProjection
from pynamodb.models import Model
classStatusIndex(GlobalSecondaryIndex):
classMeta:
        index_name = "idx_status_updated"
        projection = AllProjection()
        read_capacity_units = 3
        write_capacity_units = 2
    status = UnicodeAttribute(hash_key=True)
    updated_at = UTCDateTimeAttribute(range_key=True)
classOrderState(Model):
classMeta:
        table_name = os.getenv("ORDER_TABLE", "order_state")
        region = os.getenv("AWS_REGION", "ap-southeast-1")
# 本地调试用，线上别配这个
if os.getenv("DDB_LOCAL"):
            host = "http://localhost:8000"
    user_id = UnicodeAttribute(hash_key=True)
    order_id = UnicodeAttribute(range_key=True)
    status = UnicodeAttribute()
    amount_cent = NumberAttribute(default=0)
    updated_at = UTCDateTimeAttribute()
    remark = UnicodeAttribute(null=True)
    status_index = StatusIndex()

这段代码干了几件事。

user_id 是分区键，order_id 是排序键。这个设计我一般用在“查某个用户的订单”这种场景。别一上来就想着全表查订单，DynamoDB 不是 MySQL，建模时不把查询路径想清楚，后面只能靠 scan 硬扛。

StatusIndex 是一个 GSI，用来查某个状态下最近变化的订单，比如补偿任务扫 PAYING 超时订单。PynamoDB 支持模型、属性、索引这些抽象，也支持事务、属性序列化和反序列化这类 DynamoDB 常用能力。

建表也不用再手写一堆 JSON。

defensure_table():
if OrderState.exists():
return
    OrderState.create_table(
        read_capacity_units=5,
        write_capacity_units=5,
        wait=True,
    )

这段我一般只放在本地环境或者测试环境。线上表结构最好还是走 Terraform、CloudFormation 或 CDK，不要让应用启动时偷偷建表。这个习惯得有，不然后面权限一收紧，启动日志里全是莫名其妙的建表失败。

写入一条订单状态，看起来就清爽多了。

defcreate_order_state(user_id: str, order_id: str, amount_cent: int):
    item = OrderState(
        user_id=user_id,
        order_id=order_id,
        status="CREATED",
        amount_cent=amount_cent,
        updated_at=datetime.now(timezone.utc),
        remark=None,
    )
    item.save(
        condition=(OrderState.user_id.does_not_exist())
    )

这里的 condition 我会尽量加。

别觉得创建订单时不会重复。消息重试、接口超时、前端连点、网关重放，哪个都可能把同一笔订单打进来第二次。没有条件写，第二次就把第一次覆盖了，日志还挺安静。

状态流转也一样，不能直接改。

defmark_paid(user_id: str, order_id: str):
    order = OrderState.get(user_id, order_id)
if order.status != "CREATED":
raise RuntimeError(f"bad order status: {order.status}")
    order.update(
        actions=[
            OrderState.status.set("PAID"),
            OrderState.updated_at.set(datetime.now(timezone.utc)),
            OrderState.remark.set("paid callback accepted"),
        ],
        condition=(OrderState.status == "CREATED")
    )

这里我故意写了两层判断。

第一层在业务里挡一下，日志好看，问题也好查。

第二层 condition 才是真正防并发的。两个支付回调同时进来，谁先更新成功谁算数，后面的会失败。别只靠 Python 里的 if，那东西管不住并发。

查数据也比较像正常 Python 代码。

deflist_user_orders(user_id: str, limit: int = 20):
    rows = OrderState.query(
        user_id,
        scan_index_forward=False,
        limit=limit,
    )
return [
        {
"order_id": row.order_id,
"status": row.status,
"amount_cent": row.amount_cent,
"updated_at": row.updated_at.isoformat(),
        }
for row in rows
    ]

这里有个坑，query 和 scan 不是一回事。

query 是按分区键走，能接受。scan 是扫表，开发环境几百条数据看不出来，线上数据一多，账单和延迟会一起给你脸色看。我见过有人写后台列表图省事直接 scan，刚开始挺爽，后面数据涨起来，接口慢得像在等远程文件下载。

用 GSI 扫待处理订单，可以这么写：

defload_timeout_paying_orders(before_time, limit=100):
    items = OrderState.status_index.query(
"PAYING",
        OrderState.updated_at < before_time,
        scan_index_forward=True,
        limit=limit,
    )
for item in items:
yield item

这个写法适合补偿任务。比如支付中状态超过一段时间没结果，就拉出来查第三方支付状态。

但这里也别乱来。PAYING 如果是超级热点状态，所有订单都堆在一个分区键上，GSI 也救不了。真有这种量，要把状态拆得更细，比如加业务线、日期桶，甚至重新设计 key。

我喜欢 pynamodb 的地方，不是它让 DynamoDB 变成了关系数据库。它没这个本事，也不该这么干。

它真正舒服的地方是：把字段、索引、条件更新这些东西收回到模型里，业务代码不用满屏 boto3 参数。尤其是 Python 项目里写一些订单状态、任务状态、幂等记录、接口回调流水，pynamodb 很顺手。

但有几句话得放前面。

第一，不要拿它当 SQLAlchemy 用。DynamoDB 的分区键、排序键、GSI，还是要自己设计。

第二，不要偷懒 scan。能 query 就 query，不能 query 就回去改模型。

第三，条件更新要舍得写。状态机、幂等表、库存扣减，这些地方不写 condition，迟早会有一笔数据教你做人。

第四，本地可以用 DynamoDB Local 跑流程，线上表结构别让应用临时创建。

pynamodb 神奇的点就在这里：它不改变 DynamoDB 的规则，只是让 Python 代码少一点噪音。

少一点噪音，排障时就能多看一眼真正的问题。