Wednesday, 24 March 2021

Postgres/SqlAlchemy select record with json array contains at least 1 json from another json array

I've got the following schema:

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"}]');

Now, from this schema, I'm trying to get the records, where room_type_occupancy_mappings json array contains at least one json from another json array.

In this case, let's say I've a json array:

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

I need all the records, which contains one or all of these jsons, in the array. This I'm able to do with this query:

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)));

Problem I'm facing is in converting this query to SQLAlchemy. I tried to do something like this:

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)]))))

But this results in an error. I tried this query too:

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())

But this also results in an error.

Is there a way to convert this query to SqlAlchemy? Or should I better convert this JSONB type, to a separate table. Right now, RoomTypeOccupancyMapping is a value object in my entity modelling.



from Postgres/SqlAlchemy select record with json array contains at least 1 json from another json array

No comments:

Post a Comment