如何在GraphQL中使用PynamoDB条件表达式？解决多unit查询报错

为什么会报错？ #

DynamoDB的Query操作对**键条件表达式（KeyConditionExpression）**有严格限制：针对排序键（包括LSI的排序键），仅支持EQ、LE、LT、GE、GT、BEGINS_WITH、BETWEEN这些运算符，IN是不允许用于键条件的——这就是你收到ValidationException的直接原因。

不使用循环实现多unit查询的方案 #

方案1：使用BatchGetItem（推荐） #

如果已知目标item的完整主键（user_id+unit），可以用BatchGet批量查询，这是DynamoDB官方推荐的批量获取方式，避免循环多次查询。

代码示例：

from pynamodb.models import BatchGet

# 构造要查询的主键列表
target_keys = [
    userModel(user_id, unit=1),
    userModel(user_id, unit=2)
]

# 执行批量查询
batch_result = BatchGet().add(target_keys).execute()

# 处理结果
for item in batch_result.get(userModel, []):
    print(item)

注意：BatchGet单次最多查询100个item，总数据量不能超过16MB，超出的话需要分页处理。

方案2：使用BETWEEN（仅适用于连续unit值） #

如果你的unit值是连续范围（比如1到2），可以改用between运算符，这符合Query的键条件要求：

result = userModel.unit_index.query(
    user_id,
    userModel.unit.between(1, 2)
)

但如果unit是离散的非连续值（比如1、3、5），这个方案不适用。

不推荐：使用FilterExpression #

你可能会想到用FilterExpression来过滤unit值，但这是Query之后的过滤操作——DynamoDB会先把该user_id下所有的item都查出来，再过滤出符合unit列表的结果，当该用户下数据量大时，性能极差且浪费读写容量，因此不推荐。

GraphQL中结合PynamoDB实现多unit查询 #

以常见的Graphene框架为例，你可以这样实现：

1. 定义GraphQL查询类型 #

type User {
    user_id: String!
    unit: Int!
    # 其他字段...
}

type Query {
    getUserUnits(user_id: String!, units: [Int!]!): [User!]!
}

2. 实现Resolver函数 #

from graphene import ObjectType, String, List, Int
from pynamodb.models import BatchGet
from your_models import userModel

class Query(ObjectType):
    get_user_units = List(UserType, user_id=String(required=True), units=List(Int, required=True))

    def resolve_get_user_units(self, info, user_id, units):
        # 构造批量查询的主键列表
        target_keys = [userModel(user_id, unit=unit) for unit in units]
        # 执行批量查询
        batch_result = BatchGet().add(target_keys).execute()
        # 转换为GraphQL返回的User类型
        return [UserType(**item.attribute_values) for item in batch_result.get(userModel, [])]

如果使用AWS AppSync，你可以在Resolver的请求模板中构造BatchGetItem的操作，直接从DynamoDB批量获取数据，避免在代码层处理。

内容的提问来源于stack exchange，提问作者Darshan theerth