Filtering by JSON type field with Postgres and SQLAlchemy

Question

I have a table test_table:

 Column     |          Type          |
------------+------------------------+
 id         | integer                |
 attributes | json                   |

With content:

 id |    attributes
----+----------------------------
  1 | {"a": 1, "b": ["b1","b2"]}
  2 | {"a": 2, "b": ["b3"]}
  3 | {"a": 3}

And I need to filter the data by attribute b in the field attributes.
I found a solution using the like method.

SELECT * FROM test_table
WHERE attributes ->> 'b' SIMILAR TO '%(b1|b3)%';
-- or using SQLAlchemy
arr = ["b1", "b3"]
arr = [f"%{i}%" for i in arr]
stmt = select(test_table).where(cast(t.c.attributes["b"], String).like(any_(arr)))

The result is:

 id |    attributes
----+----------------------------
  1 | {"a": 1, "b": ["b1","b2"]}
  2 | {"a": 2, "b": ["b3"]}

But I am still trying to find a solution like this

SELECT * FROM test_table 
WHERE attributes -> 'b' ?| array['b1', 'b3'];

Is this possible with pure SQLAlchemy?

Postgres 9.6
SQLAlchemy 1.4

Answer 1

Is this possible with pure SQLAlchemy?

It is, pretty much exactly how you pictured it. Cast and use jsonb containment: demo at db<>fiddle

SELECT * FROM test_table 
WHERE attributes::jsonb -> 'b' ?| array['b1', 'b3'];

In pure SQLAlchemy you can cast() just the same, and ?| operator translates to .has_any()

arr = ["b1", "b3"]
stmt = select(test_table).where(cast(t.c.attributes["b"],JSONB).has_any(arr))

There's a table with other translations.

JSONPath would be more flexible, but while SQLAlchemy 1.4 does support it, your PostgreSQL 9.6 does not - you would need 12.0 or higher.

Answer 2

In SQLAlchemy, working with JSON data can be a bit tricky, especially when it comes to querying based on conditions within JSON fields.

If you're using SQLAlchemy 1.4 or later, you can accomplish this as follows:

from sqlalchemy import select, func
from sqlalchemy.dialects.postgresql import JSONB

# Assuming test_table is your Table object and 'attributes' is the column
stmt = select(test_table).where(
    func.jsonb_exists_any(
        cast(test_table.c.attributes['b'], JSONB),
        array(['b1', 'b3'])
    )
)