如何获取类似SQL IN子句的项目列表？附DynamoDB表定义

首先得明确：DynamoDB不像SQL那样有直接的IN子句，但根据你要查询的属性类型（主键还是非主键），咱们有几种高效的替代方案，结合你的表结构（哈希键是orgId，项目名在组织内唯一），我给你分场景说清楚：

1. 批量获取特定主键对应的项目（最推荐） #

你的表中orgId+name是唯一组合（因为项目名在组织内唯一），如果要获取多个(orgId, name)对应的项目，直接用DynamoDB的BatchGetItem操作，这是专门为批量主键查询设计的，性能最优。

举个Python boto3的示例：

import boto3

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('symbols')

# 构造要查询的主键列表
target_items = [
    {'orgId': 'org_001', 'name': 'project_A'},
    {'orgId': 'org_002', 'name': 'project_B'},
    {'orgId': 'org_001', 'name': 'project_C'}
]

response = table.batch_get_item(
    RequestItems={
        'symbols': {
            'Keys': target_items
        }
    }
)

# 获取结果
result_items = response['Responses']['symbols']

2. 按非主键属性（比如id）实现IN查询 #

如果要根据自动生成的uuid（也就是id字段）做类似IN的查询，分两种情况：

情况A：可以创建全局二级索引（GSI） #

如果经常需要按id查询，强烈建议给id建一个GSI，把id设为哈希键。这样就能用BatchGetItem批量查询，效率很高。

创建GSI的表定义补充（可以在控制台或通过API修改表）：

"GlobalSecondaryIndexes": [
    {
        "IndexName": "id-index",
        "KeySchema": [{"AttributeName": "id", "KeyType": "HASH"}],
        "Projection": {"ProjectionType": "ALL"},
        "ProvisionedThroughput": {"ReadCapacityUnits": 5, "WriteCapacityUnits": 5}
    }
]

之后用BatchGetItem查询多个id：

response = table.batch_get_item(
    RequestItems={
        'symbols': {
            'Keys': [{'id': 'uuid_001'}, {'id': 'uuid_002'}, {'id': 'uuid_003'}],
            'IndexName': 'id-index'
        }
    }
)

情况B：无法创建索引（临时查询） #

如果只是偶尔查询，不想改表结构，可以用Scan配合FilterExpression的is_in方法，但要注意：这种方式是先扫描全表再过滤，会消耗较多读取容量，数据量大时性能很差，不推荐高频使用。

示例：

from boto3.dynamodb.conditions import Attr

response = table.scan(
    FilterExpression=Attr('id').is_in(['uuid_001', 'uuid_002', 'uuid_003'])
)
result_items = response['Items']

3. 查询多个组织（orgId IN [...]）下的所有项目 #

如果要获取多个orgId对应的所有项目，DynamoDB的Query只能针对单个哈希键，所以咱们可以并行执行多个Query（每个orgId对应一个），然后合并结果，这样比扫描全表高效得多。

Python示例（用线程池并行查询）：

from concurrent.futures import ThreadPoolExecutor
from boto3.dynamodb.conditions import Key

def query_single_org(org_id):
    response = table.query(KeyConditionExpression=Key('orgId').eq(org_id))
    return response['Items']

# 要查询的组织ID列表
target_orgs = ['org_001', 'org_002', 'org_003']

# 并行执行查询
with ThreadPoolExecutor(max_workers=len(target_orgs)) as executor:
    results = executor.map(query_single_org, target_orgs)

# 合并所有结果
all_projects = []
for project_list in results:
    all_projects.extend(project_list)

内容的提问来源于stack exchange，提问作者Joe P