Python实用工具：PynamoDB——轻松操作AWS DynamoDB的ORM库

作者：

parklam

在

Python

一、PynamoDB 核心概述

1.1 用途

PynamoDB是一款为AWS DynamoDB打造的Python ORM（对象关系映射）库，能让开发者以面向对象的方式定义数据表结构、执行增删改查操作，无需编写复杂的原生API调用代码，大幅降低DynamoDB的使用门槛。

1.2 工作原理

PynamoDB将Python类映射为DynamoDB数据表，类的属性对应数据表的字段，通过封装AWS SDK for Python（boto3）的底层接口，把对象的实例化、方法调用转化为DynamoDB的API请求，实现数据表的创建、数据的读写等操作。

1.3 优缺点

优点：语法简洁直观，符合Python开发者的使用习惯；支持自动分页、条件查询、事务操作等高级功能；兼容DynamoDB的本地测试环境，便于开发调试。
缺点：相比原生boto3，存在轻微的性能损耗；部分DynamoDB的冷门特性支持不够及时；依赖AWS账号配置，本地开发需额外配置环境。

1.4 License类型

PynamoDB采用MIT License开源协议，允许开发者自由使用、修改、分发代码，无论是商业项目还是开源项目都能无门槛集成。

二、PynamoDB 安装与环境配置

2.1 安装命令

PynamoDB已发布至PyPI，可直接通过pip包管理器安装，建议安装最新稳定版本：

pip install pynamodb

若需要使用最新的开发特性，也可以从GitHub源码安装：

pip install git+https://github.com/pynamodb/PynamoDB.git

2.2 环境配置

使用PynamoDB操作AWS DynamoDB前，需要配置AWS账号的访问凭证，主要有两种方式：

配置环境变量
在系统中设置以下两个环境变量，分别对应AWS的Access Key ID和Secret Access Key：

export AWS_ACCESS_KEY_ID="your_access_key_id"
export AWS_SECRET_ACCESS_KEY="your_secret_access_key"
export AWS_DEFAULT_REGION="us-east-1"  # 例如：us-east-1、ap-southeast-1

使用AWS配置文件
在用户目录下创建.aws文件夹，新建credentials和config两个文件：

credentials文件内容：

[default]
aws_access_key_id = your_access_key_id
aws_secret_access_key = your_secret_access_key

config文件内容：

[default]
region = us-east-1

本地测试环境配置
如果没有AWS账号，可使用DynamoDB Local进行本地开发测试。首先下载并启动DynamoDB Local，然后在代码中指定host参数连接本地服务：

# 启动DynamoDB Local（需提前安装Java）
java -Djava.library.path=./DynamoDBLocal_lib -jar DynamoDBLocal.jar -sharedDb

三、PynamoDB 核心使用方法

3.1 定义数据表模型

PynamoDB的核心是通过Python类定义数据表结构，每个类对应一个DynamoDB表，类属性对应表的键（分区键、排序键）和普通属性。

3.1.1 基础表模型定义

以下示例定义一个存储用户信息的数据表，包含分区键user_id，普通属性username、age、email：

from pynamodb.models import Model
from pynamodb.attributes import UnicodeAttribute, NumberAttribute

class UserModel(Model):
    """
    定义DynamoDB用户数据表模型
    """
    class Meta:
        # 数据表名称
        table_name = "user_table"
        # AWS区域
        region = "us-east-1"
        # 本地测试时取消注释，指定DynamoDB Local地址
        # host = "http://localhost:8000"

    # 分区键：用户ID，字符串类型
    user_id = UnicodeAttribute(hash_key=True)
    # 普通属性：用户名，字符串类型
    username = UnicodeAttribute()
    # 普通属性：年龄，数字类型
    age = NumberAttribute()
    # 普通属性：邮箱，字符串类型，可选
    email = UnicodeAttribute(null=True)

代码说明：

继承pynamodb.models.Model类是定义数据表的前提；
Meta类用于配置表的元数据，包括表名、区域、连接地址等；
UnicodeAttribute和NumberAttribute是PynamoDB提供的属性类型，分别对应字符串和数字类型，支持的属性类型还有BooleanAttribute、BinaryAttribute等；
hash_key=True表示该属性为分区键，若需要排序键，可设置range_key=True。

3.1.2 包含排序键的表模型定义

如果数据表需要复合主键（分区键+排序键），可添加一个属性并设置range_key=True，以下示例定义一个存储用户订单的数据表：

from pynamodb.models import Model
from pynamodb.attributes import UnicodeAttribute, NumberAttribute, UTCDateTimeAttribute
import datetime

class UserOrderModel(Model):
    """
    定义用户订单数据表模型（复合主键：user_id + order_id）
    """
    class Meta:
        table_name = "user_order_table"
        region = "us-east-1"
        # host = "http://localhost:8000"

    # 分区键：用户ID
    user_id = UnicodeAttribute(hash_key=True)
    # 排序键：订单ID
    order_id = UnicodeAttribute(range_key=True)
    # 订单金额
    amount = NumberAttribute()
    # 订单创建时间
    create_time = UTCDateTimeAttribute(default=datetime.datetime.utcnow)

# 创建数据表（仅需执行一次）
if not UserOrderModel.exists():
    UserOrderModel.create_table(read_capacity_units=1, write_capacity_units=1, wait=True)

代码说明：

UTCDateTimeAttribute用于存储UTC时间，default参数可设置默认值为当前时间；
exists()方法判断表是否存在，create_table()方法用于创建表，read_capacity_units和write_capacity_units分别设置读写容量单位，wait=True表示等待表创建完成。

3.2 数据增删改查操作

定义好表模型后，就可以通过实例化模型类来执行数据的增删改查操作。

3.2.1 新增数据

使用模型类的构造方法创建实例，调用save()方法将数据存入DynamoDB：

# 新增用户数据
user = UserModel(
    user_id="u001",
    username="zhangsan",
    age=25,
    email="[email protected]"
)
user.save()
print(f"新增用户：{user.username}")

# 新增订单数据
order = UserOrderModel(
    user_id="u001",
    order_id="order001",
    amount=99.9
)
order.save()
print(f"新增订单：{order.order_id}，金额：{order.amount}")

代码说明：

实例化模型类时，必须传入分区键（和排序键，若有）的参数；
save()方法会将实例数据写入对应的数据表，若主键已存在，则会覆盖原有数据。

3.2.2 查询数据

PynamoDB支持多种查询方式，包括按主键查询、条件查询、扫描表等。

（1）按主键查询单条数据

使用get()方法根据主键查询单条数据，适用于精确查询：

# 查询用户ID为u001的用户
try:
    user = UserModel.get("u001")
    print(f"用户ID：{user.user_id}")
    print(f"用户名：{user.username}")
    print(f"年龄：{user.age}")
    print(f"邮箱：{user.email}")
except UserModel.DoesNotExist:
    print("用户不存在")

# 查询用户u001的订单order001
try:
    order = UserOrderModel.get("u001", "order001")
    print(f"订单ID：{order.order_id}")
    print(f"金额：{order.amount}")
    print(f"创建时间：{order.create_time}")
except UserOrderModel.DoesNotExist:
    print("订单不存在")

代码说明：

get()方法的参数顺序为分区键、排序键（若有）；
如果查询的数据不存在，会抛出DoesNotExist异常，需要捕获处理。

（2）条件查询多条数据

使用query()方法进行条件查询，适用于按分区键查询并添加过滤条件，以下示例查询用户u001的所有订单：

# 查询用户u001的所有订单
orders = UserOrderModel.query("u001")
for order in orders:
    print(f"订单ID：{order.order_id}，金额：{order.amount}")

# 查询用户u001金额大于50的订单
from pynamodb.conditions import GreaterThan
orders = UserOrderModel.query(
    "u001",
    UserOrderModel.amount > GreaterThan(50)
)
for order in orders:
    print(f"符合条件的订单：{order.order_id}，金额：{order.amount}")

代码说明：

query()方法默认按分区键查询，返回该分区键下的所有数据；
可以通过条件表达式添加过滤条件，pynamodb.conditions提供了GreaterThan、LessThan、Contains等多种条件判断类。

（3）扫描整张表

使用scan()方法扫描整张表，返回所有数据（注意：DynamoDB扫描操作消耗容量较大，不建议在大表中使用）：

# 扫描所有用户数据
users = UserModel.scan()
for user in users:
    print(f"用户：{user.username}，年龄：{user.age}")

# 扫描年龄大于20的用户
users = UserModel.scan(UserModel.age > 20)
for user in users:
    print(f"年龄大于20的用户：{user.username}")

3.2.3 更新数据

更新数据有两种方式：一是获取数据实例后修改属性，调用save()方法；二是使用update()方法直接更新，支持条件更新。

（1）实例更新

# 获取用户实例并更新年龄
user = UserModel.get("u001")
user.age = 26
user.save()
print(f"更新后年龄：{user.age}")

（2）直接更新（推荐）

# 直接更新用户u001的邮箱，支持条件更新
from pynamodb.conditions import Attr
UserModel.update(
    "u001",
    actions=[
        UserModel.email.set("[email protected]")
    ],
    condition=Attr("age").exists()  # 条件：age属性存在
)
print("邮箱更新成功")

代码说明：

update()方法的actions参数指定要执行的更新操作，支持set、add、delete等动作；
condition参数设置更新的条件，只有满足条件时才会执行更新。

3.2.4 删除数据

使用delete()方法删除指定主键的数据，支持条件删除：

# 删除用户u001
UserModel.delete("u001")
print("用户删除成功")

# 条件删除订单：删除金额小于100的订单
UserOrderModel.delete(
    "u001",
    "order001",
    condition=UserOrderModel.amount < 100
)
print("订单删除成功")

3.3 高级功能使用

3.3.1 自动分页查询

当查询结果较多时，DynamoDB会自动分页，PynamoDB的查询结果对象支持迭代器，可直接遍历所有数据，无需手动处理分页：

# 查询所有订单，自动处理分页
orders = UserOrderModel.scan()
for order in orders:
    print(f"订单ID：{order.order_id}，用户ID：{order.user_id}")
# 也可以手动获取分页标记
last_evaluated_key = orders.last_evaluated_key
while last_evaluated_key:
    orders = UserOrderModel.scan(exclusive_start_key=last_evaluated_key)
    for order in orders:
        print(f"订单ID：{order.order_id}")
    last_evaluated_key = orders.last_evaluated_key

3.3.2 事务操作

PynamoDB支持事务操作，可通过TransactionWrite类执行多个增删改操作，确保事务的原子性：

from pynamodb.transactions import TransactionWrite

# 事务操作：新增用户和订单
with TransactionWrite() as transaction:
    # 新增用户
    user = UserModel(user_id="u002", username="lisi", age=30)
    transaction.save(user)
    # 新增订单
    order = UserOrderModel(user_id="u002", order_id="order002", amount=199.9)
    transaction.save(order)
print("事务执行成功，用户和订单已新增")

代码说明：

使用with语句创建事务上下文，在上下文中执行的save、update、delete操作会被纳入事务；
事务中任意一个操作失败，整个事务会回滚，所有操作都不会生效。

四、实际应用案例：用户订单管理系统

4.1 案例需求

开发一个简单的用户订单管理系统，实现以下功能：

注册新用户；
为用户创建订单；
查询指定用户的所有订单；
更新用户信息；
删除用户及对应的订单。

4.2 完整代码实现

from pynamodb.models import Model
from pynamodb.attributes import UnicodeAttribute, NumberAttribute, UTCDateTimeAttribute
from pynamodb.conditions import Attr, GreaterThan
from pynamodb.transactions import TransactionWrite
import datetime

# 定义用户表模型
class UserModel(Model):
    class Meta:
        table_name = "user_management_table"
        region = "us-east-1"
        # host = "http://localhost:8000"

    user_id = UnicodeAttribute(hash_key=True)
    username = UnicodeAttribute()
    age = NumberAttribute()
    email = UnicodeAttribute(null=True)

# 定义订单表模型
class OrderModel(Model):
    class Meta:
        table_name = "order_management_table"
        region = "us-east-1"
        # host = "http://localhost:8000"

    user_id = UnicodeAttribute(hash_key=True)
    order_id = UnicodeAttribute(range_key=True)
    amount = NumberAttribute()
    create_time = UTCDateTimeAttribute(default=datetime.datetime.utcnow)

# 创建数据表（首次运行时执行）
def create_tables():
    if not UserModel.exists():
        UserModel.create_table(read_capacity_units=2, write_capacity_units=2, wait=True)
    if not OrderModel.exists():
        OrderModel.create_table(read_capacity_units=2, write_capacity_units=2, wait=True)
    print("数据表创建成功")

# 注册新用户
def register_user(user_id, username, age, email=None):
    user = UserModel(
        user_id=user_id,
        username=username,
        age=age,
        email=email
    )
    user.save()
    print(f"用户 {username} 注册成功")

# 创建用户订单
def create_order(user_id, order_id, amount):
    order = OrderModel(
        user_id=user_id,
        order_id=order_id,
        amount=amount
    )
    order.save()
    print(f"订单 {order_id} 创建成功，金额：{amount}")

# 查询用户所有订单
def query_user_orders(user_id):
    orders = OrderModel.query(user_id)
    print(f"\n用户 {user_id} 的订单列表：")
    for order in orders:
        print(f"订单ID：{order.order_id}，金额：{order.amount}，创建时间：{order.create_time}")

# 更新用户年龄
def update_user_age(user_id, new_age):
    UserModel.update(
        user_id,
        actions=[UserModel.age.set(new_age)],
        condition=Attr("user_id").exists()
    )
    print(f"用户 {user_id} 年龄更新为 {new_age}")

# 删除用户及订单
def delete_user_and_orders(user_id):
    with TransactionWrite() as transaction:
        # 删除用户
        transaction.delete(UserModel(user_id=user_id))
        # 删除用户所有订单
        orders = OrderModel.query(user_id)
        for order in orders:
            transaction.delete(OrderModel(user_id=user_id, order_id=order.order_id))
    print(f"用户 {user_id} 及所有订单已删除")

# 主函数
if __name__ == "__main__":
    # 创建数据表
    create_tables()

    # 注册用户
    register_user("u003", "wangwu", 28, "[email protected]")

    # 创建订单
    create_order("u003", "order003", 159.9)
    create_order("u003", "order004", 299.9)

    # 查询用户订单
    query_user_orders("u003")

    # 更新用户年龄
    update_user_age("u003", 29)

    # 删除用户及订单
    # delete_user_and_orders("u003")

代码说明：

该案例封装了用户和订单的核心操作函数，结构清晰，便于复用；
使用事务操作删除用户及订单，确保数据一致性；
运行前需确保AWS凭证配置正确，或启用本地DynamoDB环境。

五、相关资源地址

PyPI地址：https://pypi.org/project/PynamoDB
GitHub地址：https://github.com/pynamodb/PynamoDB
官方文档地址：https://pynamodb.readthedocs.io/en/latest/

关注我，每天分享一个实用的Python自动化工具。

实用工具

Python实用工具：PynamoDB——轻松操作AWS DynamoDB的ORM库

一、PynamoDB 核心概述

1.1 用途

1.2 工作原理

1.3 优缺点

1.4 License类型

二、PynamoDB 安装与环境配置

2.1 安装命令

2.2 环境配置

三、PynamoDB 核心使用方法

3.1 定义数据表模型

3.1.1 基础表模型定义

3.1.2 包含排序键的表模型定义

3.2 数据增删改查操作

3.2.1 新增数据

3.2.2 查询数据

（1）按主键查询单条数据

（2）条件查询多条数据

（3）扫描整张表

3.2.3 更新数据

（1）实例更新

（2）直接更新（推荐）

3.2.4 删除数据

3.3 高级功能使用

3.3.1 自动分页查询

3.3.2 事务操作

四、实际应用案例：用户订单管理系统

4.1 案例需求

4.2 完整代码实现

五、相关资源地址

更多文章

Python实用工具Beam：轻量化任务调度与异步执行入门教程

Python实用工具：Celery分布式任务队列入门与实战教程

Python实用工具：Apache Airflow 从入门到实战保姆级教程

Python Squirrel库入门教程：高效数据缓存与持久化工具

Python实用工具：PynamoDB——轻松操作AWS DynamoDB的ORM库

一、PynamoDB 核心概述

1.1 用途

1.2 工作原理

1.3 优缺点

1.4 License类型

二、PynamoDB 安装与环境配置

2.1 安装命令

2.2 环境配置

三、PynamoDB 核心使用方法

3.1 定义数据表模型

3.1.1 基础表模型定义

3.1.2 包含排序键的表模型定义

3.2 数据增删改查操作

3.2.1 新增数据

3.2.2 查询数据

（1）按主键查询单条数据

（2）条件查询多条数据

（3）扫描整张表

3.2.3 更新数据

（1）实例更新

（2）直接更新（推荐）

3.2.4 删除数据

3.3 高级功能使用

3.3.1 自动分页查询

3.3.2 事务操作

四、实际应用案例：用户订单管理系统

4.1 案例需求

4.2 完整代码实现

五、相关资源地址

更多文章

Python实用工具Beam：轻量化任务调度与异步执行入门教程

Python实用工具：Celery分布式任务队列入门与实战教程

Python实用工具：Apache Airflow 从入门到实战 保姆级教程

Python Squirrel库入门教程：高效数据缓存与持久化工具

Python实用工具：Apache Airflow 从入门到实战保姆级教程