How to use JSON_EACH / JSONB_EACH in sqlalchemy

test_json_each.py

import pytest

from sqlalchemy.exc import SAWarning
from sqlalchemy import Column, func, create_engine, Integer, JSON
from sqlalchemy.orm import declarative_base, Session
Base = declarative_base()

class EventHistory(Base):
    __tablename__ = 'event_history'
    id = Column(Integer, nullable=False, primary_key=True)
    event = Column(JSON, nullable=False)

e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)


s = Session(e)
s.add_all(EventHistory(id=ix, event=e) for ix,e in enumerate([
    {"about": ["antenna", "connection", "modem", "network"]},
    {"about": ["connection", "network"]},
    {"about": ["other", "keywords"]}]))
s.commit()


def assert_results_ok(events, issue_warning=False):
    with pytest.warns(None) as record:
        result = events.order_by(EventHistory.id).limit(3).all()
        assert len(result) == 2
        assert result[0].id == 0
        assert result[1].id == 1

    if issue_warning:
        assert len(record) == 1
        assert 'cartesian product between' in str(record[0].message)
    else:
        assert list(r.message for r in record) == []


def test_ok_column():
    from sqlalchemy import column

    about_exp = func.json_each(EventHistory.event, '$.about')
    value = column('value')
    events = s.query(EventHistory).select_from(about_exp).filter(value == 'connection')
    assert_results_ok(events, issue_warning=False)  # for some odd reason it does not issue a warning
     

def test_ok_literal_column():
    from sqlalchemy import literal_column

    about_exp = func.json_each(EventHistory.event, '$.about').alias('b')
    value = literal_column('b.value')
    events = s.query(EventHistory).select_from(about_exp).filter(value ==  'connection')
    assert_results_ok(events, issue_warning=True)


def test_ok_column_valued():
    about_exp = func.json_each(EventHistory.event, '$.about').column_valued('value')
    events = s.query(EventHistory).filter(about_exp ==  'connection')
    assert_results_ok(events, issue_warning=True)


def test_ok_table_valued():
    about_exp = func.json_each(EventHistory.event, '$.about').table_valued('value')
    events = s.query(EventHistory).filter(about_exp.c.value ==  'connection')
    assert_results_ok(events, issue_warning=True)


def test_ok_table_valued_join_with_true_func():
    from sqlalchemy import true

    about_exp = func.json_each(EventHistory.event, '$.about').table_valued('value')
    events = s.query(EventHistory).join(about_exp, true()).filter(about_exp.c.value ==  'connection')
    assert_results_ok(events)