elem = func.jsonb_array_elements(Test.data, type_=JSONB).column_valued("elem")
looks good:
>>> elem = func.jsonb_array_elements(Test.data, type_=JSONB).column_valued("elem")
>>> print(select(elem))
SELECT elem
FROM jsonb_array_elements(test_table_json.data) AS elem
we now need the PG "astext" operator ->>. This is part of the JSON/JSONB operators for Postgresql and is mentioned here, so we can get that like:
>>> print(select(elem['code'].astext))
SELECT elem ->> :elem_1 AS anon_1
FROM jsonb_array_elements(test_table_json.data) AS elem
b. wrapping that to an ARRAY. the PG ARRAY literal is avilable from postgresql.array
>>> print(array([select(elem['code'].astext)]))
<stdin>:1: SAWarning: implicitly coercing SELECT object to scalar subquery; please use the .scalar_subquery() method to produce a scalar subquery.
ARRAY[(SELECT elem ->> %(elem_1)s AS anon_1
FROM jsonb_array_elements(test_table_json.data) AS elem)]
OK lets do what the warning says:
>>> print(array([select(elem['code'].astext).scalar_subquery()]))
ARRAY[(SELECT elem ->> %(elem_1)s AS anon_1
FROM jsonb_array_elements(test_table_json.data) AS elem)]
c. left hand literal expressions with the "::" operator, that's SQL CAST. we can use CAST directly to get this (found an issue in my tutorial, that the current CAST tutorial is in the old tutorial only at the moment, that needs to be migrated):
>>> print(cast("example code", ARRAY(Text)).compile(dialect=postgresql.dialect()))
CAST(%(param_1)s::TEXT[] AS TEXT[])
d. the docs aren't good on this one. The PG <@ operator. This is called "contained by", and it's here: https://docs.sqlalchemy.org/en/14/dialects/postgresql.html?highlight=jsonb#sqlalchemy.dialects.postgresql.ARRAY.Comparator.contained_by however this doc should show the operator. anyway it looks like this:
>>> print(
... cast("example code", ARRAY(Text)).
... contained_by(array([select(elem['code'].astext).scalar_subquery()]))
... )
CAST(:param_1 AS ARRAY) <@ ARRAY[(SELECT elem ->> %(elem_1)s AS anon_1
FROM jsonb_array_elements(test_table_json.data) AS elem)]
e. OK the whole thing, which also has the two-level JSONB element and the OR, see below
from sqlalchemy import Column
from sqlalchemy import func
from sqlalchemy import Integer
from sqlalchemy import or_
from sqlalchemy import select
from sqlalchemy import Text, cast
from sqlalchemy.dialects.postgresql import ARRAY
from sqlalchemy.dialects.postgresql import array
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class Test(Base):
__tablename__ = 'test_table_json'
id = Column(Integer, primary_key=True)
data = Column(JSONB)
elem = func.jsonb_array_elements(Test.data, type_=JSONB).column_valued("elem")
stmt = select(Test).where(
or_(
cast("example code", ARRAY(Text)).contained_by(
array([select(elem['code'].astext).scalar_subquery()])
cast("stefan", ARRAY(Text)).contained_by(
array([select(elem['code']['new_value'].astext).scalar_subquery()])
print(stmt)
this one had a lot of special APIs to use, so for now I'd advise just going through https://docs.sqlalchemy.org/en/14/dialects/postgresql.html to see what's available.
@zzzeek, it doesn't work as expected.
bad contained_by(array([select(elem['code'].astext).scalar_subquery()
Working RAW:
Generated RAW:
additional square quotes are created after ARRAY, although they are not actually needed.
cast("example code", ARRAY(Text))
Generated param: ['s', 't', 'e', 'f', 'a', 'n']
When expected: '{stefan}'
you can solve it like this, but it seems to me that this is not the best option:
cast(text("'{stefan}'"), ARRAY(Text))
oh right array() literal doesn't do subqueries.
just use cast() as needed and for the parameters, those should be lists since that's array.
stmt = select(Test).where(
or_(
cast(["example code"], ARRAY(Text)).contained_by(
array([select(elem['code'].astext).scalar_subquery()])
cast(["stefan"], ARRAY(Text)).contained_by(
cast(select(elem['code']['new_value'].astext).scalar_subquery(), ARRAY(Text))
stmt = select(Test).where(
or_(
cast(["example code"], ARRAY(Text)).contained_by(
array([select(elem['code'].astext).scalar_subquery()])
cast(["stefan"], ARRAY(Text)).contained_by(
func.array(select(elem['code']['new_value'].astext).scalar_subquery())