如何将SqlAlchemy查询结果转换为嵌套字典（Python）

需求说明 #

查询结果 #

| device_id |     event_name      | event_count |
| :--------:| :-----------------: |:-----------:|
| 123456    | speed more than 100 |      3      |
| 123456    | speed less than 12  |      0      |
| 334455    | out of NYC          |      5      |
| 111111    | in UCSD campus      |      1      |

目标嵌套字典格式 #

{'123456' : {'speed more than 100' : 3,
             'speed less than 12': 0},
 '334455' : {'out of NYC' : 5},
 '111111' : {'in UCSD campus' : 1}
}

现有代码 #

未完成的转换函数 #

def count_per_event_json(self, count_per_event_query_result):
        result = {}
        print(count_per_event_query_result)
        for item in enumerate(count_per_event_query_result):
            if item[0] not in result.keys():
                # result[item[0]] = {I don't know how to fill this inner dict'}
    
        return result

查询逻辑代码 #

def count_per_event(self):
    count_per_event_query = select(EventsModel.device_id, EventsModel.event_name,
                                   func.count(EventsModel.rule_table_id)) \
        .where(EventsModel.timestamp <= self.max_utc_timestamp) \
        .where(EventsModel.timestamp >= self.min_utc_timestamp) \
        .group_by(EventsModel.device_id, EventsModel.rule_table_id) \
        .execution_options(synchronize_session="fetch")

    return count_per_event_query


async def cube_calculator(self):
     async with async_session() as session:
                count_per_event_query_result = await session.execute(self.count_per_event())
                json = self.count_per_event_json(count_per_event_query_result) # 类型为 <class 'sqlalchemy.engine.result.ChunkedIteratorResult'>

解决方案 #

步骤1：给查询结果的计数字段添加别名 #

先修改count_per_event函数，为func.count的结果添加别名，方便后续通过属性访问：

def count_per_event(self):
    count_per_event_query = select(EventsModel.device_id, EventsModel.event_name,
                                   # 添加别名event_count
                                   func.count(EventsModel.rule_table_id).label('event_count')) \
        .where(EventsModel.timestamp <= self.max_utc_timestamp) \
        .where(EventsModel.timestamp >= self.min_utc_timestamp) \
        .group_by(EventsModel.device_id, EventsModel.rule_table_id) \
        .execution_options(synchronize_session="fetch")

    return count_per_event_query

步骤2：完成转换函数的核心逻辑 #

修改count_per_event_json函数，正确遍历查询结果并构建嵌套字典：

def count_per_event_json(self, count_per_event_query_result):
    result = {}
    # 从异步查询结果中获取所有行数据
    rows = count_per_event_query_result.all()
    
    for row in rows:
        # 将device_id转为字符串，与目标格式一致
        device_id = str(row.device_id)
        event_name = row.event_name
        event_count = row.event_count
        
        # 如果当前device_id不在结果字典中，初始化内层字典
        if device_id not in result:
            result[device_id] = {}
        # 将事件名称和计数映射到内层字典
        result[device_id][event_name] = event_count
    
    return result

关键说明 #

• 移除了错误的enumerate：原代码中enumerate返回的是循环索引和行对象，item[0]并非device_id，这是核心错误点。
• 处理异步查询结果：ChunkedIteratorResult需要调用all()方法获取所有行数据，才能正常遍历处理。
• 统一device_id类型：转为字符串确保与目标字典的键类型完全匹配。

内容的提问来源于stack exchange，提问作者Aylin Naebzadeh