时间戳

首页 > 解决方案 > 带有 json 数组的 Postgres/SqlAlchemy 选择记录包含来自另一个 json 数组的至少 1 个 json

python - 带有 json 数组的 Postgres/SqlAlchemy 选择记录包含来自另一个 json 数组的至少 1 个 json

问题描述

我有以下架构：

CREATE TABLE rate_plan (
    rate_plan_id SERIAL PRIMARY KEY,
    room_type_occupancy_mappings JSONB NOT NULL
);


INSERT INTO rate_plan 
VALUES 
    (1, '[{"adult_count": 1, "room_type_id": "RT01"}, {"adult_count": 1, "room_type_id": "RT02"}, {"adult_count": 1, "room_type_id": "RT03"}, {"adult_count": 1, "room_type_id": "RT04"}]'),
    (2, '[{"adult_count": 2, "room_type_id": "RT02"}]'),
    (3, '[{"adult_count": 1, "room_type_id": "RT02"}]');

现在，从这个模式中，我试图获取记录，其中room_type_occupancy_mappingsjson 数组包含来自另一个 json 数组的至少一个 json。

在这种情况下，假设我有一个 json 数组：

[{"room_type_id": "RT02", "adult_count": 2"}, {"room_type_id": "RT03", "adult_count": 1"}]

我需要数组中包含一个或所有这些 json 的所有记录。这我可以用这个查询来做：

SELECT rate_plan_id, room_type_occupancy_mappings 
FROM rate_plan 
WHERE EXISTS (
    SELECT 1 
    FROM jsonb_array_elements(room_type_occupancy_mappings) x 
    WHERE (x->>'room_type_id', (x->>'adult_count')::int) IN (('RT02', 2), ('RT03', 1)));

我面临的问题是将这个查询转换为 SQLAlchemy。我试图做这样的事情：

query = self.session().query(RatePlan).filter(
    exists(
        select([1]).select_from(
            func.jsonb_array_elements(self._model.room_type_occupancy_mappings)
                .filter(tuple_(self._model.room_type_occupancy_mappings['room_type_id'].astext,
                         self._model.room_type_occupancy_mappings['adult_count'].astext.cast(Integer))
                        ).in_([(‘RT02’, 2), (‘RT03’, 1)]))))

但这会导致错误。我也试过这个查询：

query = self.session().query(RatePlan).filter(
            repo.session().query(func.jsonb_array_elements(repo._model.room_type_occupancy_mappings).alias('rtom'))
                .filter(tuple_(rtom['room_type_id'].astext,
                               rtom['adult_count'].astext.cast(Integer)
                              ).in_([('RT02', 2), ('RT03', 1)])).exists())

但这也会导致错误。

有没有办法将此查询转换为 SqlAlchemy？或者我应该更好地将此 JSONB 类型转换为单独的表。现在，RoomTypeOccupancyMapping是我的实体建模中的一个值对象。

标签： pythondatabasepostgresqlsqlalchemyjsonb

解决方案

最后，我使用以下查询使其工作：

rtom_value = literal_column('rtom.value', type_=JSONB)

query = self.session().query(RatePlan).filter(
            self.session().query().select_from(func.jsonb_array_elements(self._model.room_type_occupancy_mappings).alias('rtom'))
                .filter(tuple_(rtom_value['room_type_id'].astext,
                               rtom_value['adult_count'].astext.cast(Integer)
                              ).in_([('RT02', 2), ('RT03', 1)])).exists())

推荐阅读