pynamodb，一个神奇的 Python 库！

线上表不大，代码先乱了。

同样是往 DynamoDB 写一条数据，团队里常见两种写法：一堆 boto3.client("dynamodb") 手搓字典，{"S": "...", "N": "..."} 满天飞；另一种更离谱，业务字段已经改了，序列化层还在偷偷兼容旧字段，最后排查半天，不是表有问题，是人把自己绕进去了。

PynamoDB 这东西，我第一眼就挺信。它干的不是“替你理解 DynamoDB”，而是把那层最烦的样板代码先铲掉：模型定义、属性序列化、条件更新、批量写、事务操作，基本都给你接住了。Meta 里至少要告诉它表名，区域也得配清楚；新版本里它已经不再默认给你兜一个 AWS region 了，这种细节不盯着，代码本地能跑，上线就容易翻车。

我一般不会上来就吹“它像 ORM”。真到 DynamoDB 这儿，你要是把它当 Django ORM 那样随便使，后面八成要吃亏。PynamoDB 更像一个很顺手的 Python 包装层：模型还是你自己控，分区键、排序键、索引、条件表达式这些核心东西，一样要老老实实想明白。只不过写出来没那么硌手。

先看个我更愿意在线上落的模型，不花哨，但够用：

from datetime import datetime
from pynamodb.models import Model
from pynamodb.attributes import (
    UnicodeAttribute,
    NumberAttribute,
    UTCDateTimeAttribute,
    JSONAttribute,
    VersionAttribute,
)
classOrderModel(Model):
classMeta:
        table_name = "order_center"
        region = "ap-southeast-1"
    order_id = UnicodeAttribute(hash_key=True)
    biz_type = UnicodeAttribute(range_key=True)
    status = UnicodeAttribute(default="CREATED")
    amount = NumberAttribute(default=0)
    ext = JSONAttribute(null=True)
    created_at = UTCDateTimeAttribute(default=datetime.utcnow)
    updated_at = UTCDateTimeAttribute(default=datetime.utcnow)
    version = VersionAttribute()

这个写法的好处很直接。字段类型、默认值、乐观锁，全在一个地方。后面你查单、改状态、补扩展字段，脑子里不用再来回翻“这个字段进库前是不是还要手动转一下”。PynamoDB 自带的属性类会负责 DynamoDB 属性的序列化和反序列化，像 JSONAttribute、UTCDateTimeAttribute 这种，在线上确实省事。

真正让我觉得它顺手的，是更新操作。很多人写 DynamoDB，喜欢先 get，改对象，再 save。能不能用？能。只是你一旦碰上并发更新、库存扣减、状态流转，这么写就开始发虚了。我更愿意把条件和动作一起塞进更新里，少一次心跳，少一层侥幸。

from datetime import datetime
defpay_order(order_id: str, biz_type: str, paid_amount: int):
    item = OrderModel(order_id, biz_type)
    item.update(
        actions=[
            OrderModel.status.set("PAID"),
            OrderModel.amount.set(paid_amount),
            OrderModel.updated_at.set(datetime.utcnow()),
            OrderModel.ext.set({"pay_channel": "wx", "confirmed": True}),
        ],
        condition=(
            (OrderModel.status == "CREATED") &
            (OrderModel.amount >= 0)
        )
    )

这里值钱的不是 set() 这个动作本身，而是这个味道：改数据的时候，把“只有 CREATED 才能改成 PAID”一并写进去。PynamoDB 支持用属性对象拼 update expression，也支持条件表达式；底下走的还是 DynamoDB 那套能力，但代码终于像人写的了。

再往前走一步，线上更常见的是“不是不会写，是怕写重”。比如补偿任务、重复回调、消费者重试，这些地方最烦的不是异常，而是悄悄把数据写坏。PynamoDB 有 VersionAttribute，本质就是乐观锁。读到什么版本，写回去时就校验什么版本，不一致直接失败，别装作没看见。官方文档里也明确说了，它会把版本号作为并发冲突检测的一部分。

defclose_order(order_id: str, biz_type: str):
    order = OrderModel.get(order_id, biz_type)
    order.status = "CLOSED"
    order.updated_at = datetime.utcnow()
    order.save()   # version 不对会直接抛异常

这种代码我挺喜欢。它不帮你“解决并发”，但至少不让你悄无声息地覆盖别人刚写进去的数据。很多脏数据，不是业务不会判断，是更新时太随便。

批量写这块，PynamoDB 也很实用。DynamoDB 原生批量写有上限，自己手搓分组很烦；PynamoDB 的 batch_write() 会帮你按批处理，官方文档写得很明白：批量写一次最多 25 个写请求，它会自动帮你拆。

defbatch_init_orders(rows: list[dict]):
with OrderModel.batch_write() as batch:
for row in rows:
            batch.save(
                OrderModel(
                    row["order_id"],
                    row["biz_type"],
                    status="CREATED",
                    amount=row["amount"],
                    ext={"source": row.get("source", "import")}
                )
            )

但这里我得多说一句，别一看到 batch 就上头。批量写不是批量更新，DynamoDB 的 BatchWriteItem 只支持 Put 和 Delete，不支持带条件的 Update。你要做“多条数据要么一起成功，要么一起失败”，该上事务就上事务，别在业务里自己拼半套补偿。PynamoDB 也支持事务操作，而且是用上下文管理器包起来的，这种写法比你手动调底层 API 稳很多。

所以 PynamoDB 神奇在哪？

不是因为它把 DynamoDB 变简单了。DynamoDB 该你想的建模、热分区、访问模式，它一点都没替你想。它神奇在另一层：把 Python 里那堆又臭又长、还容易写错的 DynamoDB 操作，收成了能维护的模型代码。字段是字段，条件是条件，更新动作是更新动作，读代码的人不需要先把脑子切到 AWS SDK 的报文模式。

真要我给一句不那么工整的评价：这库很适合“已经决定用 DynamoDB，但不想在 Python 里一直揉 boto3 细节”的团队。小项目用它，代码利索；线上项目用它，改字段、加条件、补乐观锁，至少不容易把自己写恶心。行，差不多就写到这。