带有连接的 SQLAlchemy 查询的嵌套 JSON 输出

Question

我有两个表 Orders 和 OrderItems。这是一种常见的设置，OrderItems 有一个将其链接到 Orders 的外键。所以我们有一个从 Orders 到 OrderItems 的一对多连接。 注意：表格在现实生活中会有更多的字段。

Orders         OrderItems
+---------+   +-------------+---------+
| orderId |   | orderItemId | orderId |
+---------+   +-------------+---------+
|       1 |   |           5 |       1 |
|       2 |   |           6 |       1 |
|         |   |           7 |       2 |
+---------+   +-------------+---------+

我正在使用 SQLAlchemy 来反映现有数据库。所以要查询这些数据，我会做类似的事情

ordersTable = db.Model.metadata.tables['Orders']
orderItemsTable = db.Model.metadata.tables['OrdersItems']
statement = ordersTable.join(orderItemsTable, ordersTable.c.orderId==orderItemsTable.c.orderId).select()
result = db.engine.execute(statement)
rlist = [dict(row) for row in result.fetchall()]

return flask.jsonify(rlist)

但是这个输出的问题是，由于连接，我从 Orders 表中获得了重复的信息。例如。您可以看到，因为 orderId 有两个项目，所以我将两次获取 Orders 表中的所有内容。

我追求的是一种从选择查询 aboce 获取嵌套 JSON 输出的方法。如：

[
    {
        "orderId": 1, 
        "orderItems": [
            { "orderItemId": 5 },
            { "orderItemId": 6 }
        ]
    },
    {
        "orderId": 2,
        "orderItems":[
            { "orderItemId": 7 }
        ]
    }
]

这个问题之前已经提出过 How do I produce nested JSON from database query with joins? Using Python / SQLAlchemy 我花了很多时间查看 Marshmallow 文档，但我找不到如何使用我上面概述的查询类型来实现它。

Answer 1

我不喜欢棉花糖有多么凌乱，所以我写了这个。我也喜欢我可以将所有数据操作保留在 SQL 语句中，而不是同时指示棉花糖做什么。

import json
from flask.json import JSONEncoder

def join_to_nested_dict(join_result):
    """
    Takes a sqlalchemy result and converts it to a dictionary.
    The models must use the dataclass decorator.
    Adds results to the right in a key named after the table the right item is contained in.
    :param List[Tuple[dataclass]] join_result:
    :return dict:
    """
    if len(join_result) == 0:
        return join_result

    # couldn't be the result of a join without two entries on each row
    assert(len(join_result[0]) >= 2)

    right_name = join_result[0][1].__tablename__
    # if there are multiple joins recurse on sub joins
    if len(join_result[0]) > 2:
        right = join_to_nested_dict([res[1:] for res in join_result])
    elif len(join_result[0]) == 2:
        right = [
            json.loads(json.dumps(row[1], cls=JSONEncoder))
            for row in join_result if row[1] is not None
        ]
    right_items = {item['id']: item for item in right}

    items = {}
    for row in join_result:
        # in the case of a right outer join
        if row[0] is None:
            continue
        if row[0].id not in items:
            items[row[0].id] = json.loads(json.dumps(row[0], cls=JSONEncoder))
        # in the case of a left outer join
        if row[1] is None:
            continue
        if right_name not in items[row[0].id]:
            items[row[0].id][right_name] = []
        items[row[0].id][right_name].append(right_items[row[1].id])
    return list(items.values())

您应该能够将结果插入此函数。但是，您需要将 dataclass decorator 添加到您的模型中，此代码才能正常工作。

statement = ordersTable.join(orderItemsTable, ordersTable.c.orderId==orderItemsTable.c.orderId).select()
result = db.engine.execute(statement)
join_to_nested_dict(result)

另外，如果你不想使用flask json 编码器，你可以删除import 和cls 参数。