How to query a list in json using SQLAlchemy and sqlite?-CodePudding

Here's my code / what I've tried. How can I query a json key containing a list?

import sqlalchemy
from sqlalchemy import Column, Integer, JSON
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Session = sessionmaker()
Base = declarative_base()


class Track(Base):  # noqa: WPS230
    __tablename__ = "track"

    id = Column(Integer, primary_key=True)
    fields = Column(JSON(none_as_null=True), default="{}")

    def __init__(self, id):
        self.id = id
        self.fields = {}


engine = sqlalchemy.create_engine("sqlite:///:memory:")
Session.configure(bind=engine)
Base.metadata.create_all(engine)  # creates tables
session = Session()

track1 = Track(id=1)
track2 = Track(id=2)
track1.fields["list"] = ["wow"]
track2.fields["list"] = ["wow", "more", "items"]

session.add(track1)
session.commit()

session.query(Track).filter(Track.fields["list"].as_string() == "wow").one()
session.query(Track).filter(Track.fields["list"].as_string() == "[wow]").one()
session.query(Track).filter(
    Track.fields["list"].as_json() == ["wow", "more", "items"]
).one()

I've also tried contains() instead of ==, but that seems to match substrings of elements as well, which I don't want..

CodePudding user response：

I managed to get the behavior I was after by utilizing json_each. To filter against the entire list, I just need to create a new json_each function for each element I want to test against.

#!/usr/bin/env python3
import sqlalchemy
from sqlalchemy import func
from sqlalchemy import Column, Integer, JSON
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Session = sessionmaker()
Base = declarative_base()


class Track(Base):
    __tablename__ = "track"

    id = Column(Integer, primary_key=True)
    fields = Column(JSON, default="{}")

    def __init__(self, id):
        self.id = id
        self.fields = {}


engine = sqlalchemy.create_engine("sqlite:///:memory:")
Session.configure(bind=engine)
Base.metadata.create_all(engine)
session = Session()

track = Track(id=1)
track.fields["list"] = ["a", "list"]

session.add(track)
session.commit()

fields_func1 = func.json_each(Track.fields, "$.list").table_valued(
    "value", joins_implicitly=True
)
fields_func2 = func.json_each(Track.fields, "$.list").table_valued(
    "value", joins_implicitly=True
)
session.query(Track).filter(fields_func1.c.value == "list").one()
session.query(Track).filter(fields_func1.c.value == "a").one()
session.query(Track).filter(
    fields_func1.c.value == "a", fields_func2.c.value == "list"
).one()