如何使用SQLAlChemy和SQLite在json中查询列表？

des4xlb0 于 2022-11-15 发布在 SQLite

关注(0)|答案(1)|浏览(239)

以下是我的代码/我尝试过的代码。如何查询包含列表的json key？

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

我还尝试了contains()而不是==，但这似乎也与元素的子字符串匹配，这是我不想要的。

sqlite

来源：https://stackoverflow.com/questions/73684853/how-to-query-a-list-in-json-using-sqlalchemy-and-sqlite

1条答案

按热度按时间

kxe2p93d1#

通过利用json_each，我设法获得了我想要的行为。要根据整个列表进行过滤，我只需要为要测试的每个元素创建一个新的json_each函数。

#!/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()

赞(0）回复(0）举报 2022-11-15

我来回答

如何使用SQLAlChemy和SQLite在json中查询列表？

1条答案

相关问题

热门标签

最新问答