pynamodb，一个神奇的 Python 库！

DynamoDB 最烦人的地方，不是慢，也不是贵，是你一写代码就像在拼一坨 JSON。

一个简单的保存订单状态，boto3 原生写法能写到人怀疑人生。字段类型要自己包，表达式要自己拼，条件更新还要注意名字冲突。代码跑是能跑，但半年后再看，第一反应一般是：这玩意儿谁敢改？

PynamoDB 就是专门治这个毛病的。

它不是把 DynamoDB 变成 MySQL，也不是让你忘掉分区键、排序键、GSI 这些东西。这个库更像是在 DynamoDB 外面套了一层 Python 写法，让表结构、字段、查询、更新看起来像正常业务代码。官方文档也把它定位成 DynamoDB 的 Pythonic 接口，用 Model 抽象去包 DynamoDB 表。

先看一段我比较愿意放进项目里的写法。

from datetime import datetime, timezone
from pynamodb.attributes import UnicodeAttribute, NumberAttribute, UTCDateTimeAttribute, VersionAttribute
from pynamodb.indexes import GlobalSecondaryIndex, AllProjection
from pynamodb.models import Model

class BuyerTimeIndex(GlobalSecondaryIndex):
    class Meta:
        index_name = "idx_buyer_created"
        projection = AllProjection()
        read_capacity_units = 5
        write_capacity_units = 2
    buyer_id = UnicodeAttribute(hash_key=True)
    created_at = UTCDateTimeAttribute(range_key=True)

class OrderSnapshot(Model):
    class Meta:
        table_name = "order_snapshot"
        region = "ap-southeast-1"
    order_id = UnicodeAttribute(hash_key=True)
    sku_id = UnicodeAttribute(range_key=True)
    buyer_id = UnicodeAttribute()
    status = UnicodeAttribute()
    pay_amount = NumberAttribute(default=0)
    created_at = UTCDateTimeAttribute(default_for_new=lambda: datetime.now(timezone.utc))
    updated_at = UTCDateTimeAttribute(null=True)
    version = VersionAttribute()
    buyer_time_index = BuyerTimeIndex()

这段代码有几个点我比较喜欢。

表结构直接贴在 Python 类上，哪个字段是 hash key，哪个字段是 range key，一眼能扫出来。GSI 也没有藏在 Terraform 或控制台里让人猜，代码里就能看到 buyer_id + created_at 是一条查询路径。

当然，DynamoDB 的老规矩还在。GSI 不是免费午餐，它有自己的读写容量，写主表时索引也要跟着更新。AWS 文档里说得很清楚，GSI 的吞吐和主表是分开的，写入表时索引也会被更新。

所以我一般不建议一上来就建一堆索引。DynamoDB 不是关系库，别拿“以后可能会查”当理由乱建 GSI。能确认查询路径，再建。

保存数据就干净多了。

def save_paid_order(event: dict) -> None:
    item = OrderSnapshot(
        event["order_id"],
        event["sku_id"],
        buyer_id=event["buyer_id"],
        status="PAID",
        pay_amount=event.get("pay_amount", 0),
    )
    item.save()

要是用 boto3 原生写，S、N、ExpressionAttributeValues 会铺满屏幕。PynamoDB 把这些脏活藏掉了，业务代码看起来终于像业务代码。

但这里别误会，PynamoDB 不是让你随便 scan 的护身符。

我见过有人把它当 ORM 用，写出这种东西：

for item in OrderSnapshot.scan(OrderSnapshot.status == "PAID"):
    handle(item)

这地方我第一眼就不太信。

小表无所谓，线上订单表一大，scan 就开始拖后腿。PynamoDB 写法再优雅，也挡不住 DynamoDB 的访问模型。该 query 就 query，该走索引就走索引。PynamoDB 的文档也明确提供 query 这层高阶 API，用来封装 DynamoDB 的查询能力。

正常一点的查法应该这样：

def list_buyer_orders(buyer_id: str, limit: int = 20):
    rows = OrderSnapshot.buyer_time_index.query(
        buyer_id,
        scan_index_forward=False,
        limit=limit,
    )
    return [
        {
            "order_id": row.order_id,
            "sku_id": row.sku_id,
            "status": row.status,
            "amount": int(row.pay_amount),
            "created_at": row.created_at.isoformat(),
        }
        for row in rows
    ]

这个接口查的是某个用户最近订单。查询路径明确：buyer_id 作为 GSI 分区键，created_at 倒序。

这才是 DynamoDB 舒服的姿势。

更新也挺顺手，尤其是条件更新。

订单状态这种东西，最怕重复回调。支付网关抖一下，MQ 重投一下，接口多进来一次，状态就可能被覆盖。这里我一般不直接 save()，而是用条件更新把边界卡死。

from pynamodb.exceptions import UpdateError

def mark_order_shipped(order_id: str, sku_id: str, express_no: str) -> bool:
    try:
        item = OrderSnapshot.get(order_id, sku_id)
        item.update(
            actions=[
                OrderSnapshot.status.set("SHIPPED"),
                OrderSnapshot.updated_at.set(datetime.now(timezone.utc)),
            ],
            condition=(OrderSnapshot.status == "PAID"),
        )
        return True
    except OrderSnapshot.DoesNotExist:
        print(f"[order-miss] order_id={order_id}, sku_id={sku_id}")
        return False
    except UpdateError as e:
        print(f"[order-skip] order_id={order_id}, sku_id={sku_id}, reason={e}")
        return False

这段代码的味道就对了。

只有 PAID 状态才能改成 SHIPPED。如果已经发货了，或者订单被取消了，这次更新直接失败。失败不是坏事，悄悄覆盖才是坏事。

还有一个我挺常用的地方：批量写。

比如每天从上游同步一批商品库存快照，不要一条一条 save()，那样网络往返会把人磨死。

def flush_stock_snapshot(rows: list[dict]) -> None:
    with OrderSnapshot.batch_write() as batch:
        for row in rows:
            batch.save(
                OrderSnapshot(
                    row["order_id"],
                    row["sku_id"],
                    buyer_id=row["buyer_id"],
                    status=row["status"],
                    pay_amount=row.get("pay_amount", 0),
                    updated_at=datetime.now(timezone.utc),
                )
            )

PynamoDB 好用，但别把它吹成万能。

它解决的是 Python 写 DynamoDB 时的“啰嗦”和“难维护”，不是解决数据建模问题。分区键设计错了，热点还是热点；查询路径没想清楚，最后还是 scan；GSI 建多了，账单照样不好看。

我对它的判断很简单：如果项目里已经选了 DynamoDB，又是 Python 技术栈，PynamoDB 值得上。尤其是 Lambda、小型后端服务、异步任务、数据同步脚本，这库用起来很顺。

但上之前要把三件事想清楚。

第一，表的主查询路径是什么。

第二，哪些字段需要条件更新。

第三，哪些地方绝对不能 scan。

这三件事没想清楚，换什么库都救不了。PynamoDB 只是让代码更像人写的，DynamoDB 那套规矩，它一点都没替你取消。

上就是“pynamodb，一个神奇的 Python 库！”的详细内容，想要了解更多Python教程欢迎持续关注编程学习网。