python - SQLAlchemy 按 PickleType 内容过滤查询

Question

我有这个 Meter 对象模型

class Meter(db.Model):
    """
    Model for repair meter.

    """
    __tablename__ = 'meters'

    id = db.Column(db.Integer, primary_key=True, nullable=False)
    meter_no = db.Column(db.String(50), unique=True, nullable=False)
    client = db.Column(db.String(50), unique=True, nullable=False)
    problems = db.Column(db.PickleType)
    functional = db.Column(db.Boolean(), default=False, server_default="false")
    location = db.Column(db.Integer, db.ForeignKey('locations.id'))
    date_in = db.Column(db.Date, default=dt.date.today())
    date_out = db.Column(db.Date)

问题 PickleType 列存储一个包含问题对象 ID 的 Python 列表

然后我有这个函数尝试查询 meters 数据库表并仅获取列表存储中具有特定问题 ID (prob_id) 的 Meter 对象作为 PickleType。

def problemrate(prob_id, month):
    """
    Return meters with the problem id for a particular month.

    args:
        prob_id (int): problem id
        month (int): month number e.g. april -> 4

    returns:
        meters (list): list of meter objects

    raises: None

    """
    meters = Meter.query.filter(and_(extract('month', Meter.date_in) == month,
                    prob_id in Meter.problems)).all()

    return meters

但是当调用该函数时出现错误:

NotImplementedError: Operator 'getitem' is not supported on this expression

通过 PickleType 列的内容过滤 sqlalchemy 查询的编写方法是什么？这是可能的还是 sqlalchemy 不支持这个？

Answer 1

简短说明:PickleType 不支持任何关系功能，例如查询/过滤器。它的目的是存储和检索。请改用 sqlalchemy.orm.relationship。

详细解释:错误信息实际上是正确的。 filter 函数中的所有内容都将编译为 sql 查询(打印查询以查看)，因此“in”运算符不会编译为有效查询，任何其他 getItem 运算符也不会。起作用的是“==”，但是为此您需要传递完全相同的对象，以便可以将其转换为与存储的对象相同的 PickleType 对象。比较在内部完成。

解决方案:在普通 SQL 中，您可以将每个列表项分别存储在不同的表中，并通过 ID 关联。在 SQLAlchemy 中你可以做类似的事情:

from sqlalchemy.orm import relationship
from sqlalchemy.ext.associationproxy import association_proxy

class Meter(db.Model):
    """
    Model for repair meter.

    """
    __tablename__ = 'meters'

    id = db.Column(db.Integer, primary_key=True, nullable=False)
    meter_no = db.Column(db.String(50), unique=True, nullable=False)
    client = db.Column(db.String(50), unique=True, nullable=False)
    meter_problems = relationship('Problem', secondary=lambda: meterproblems_table)
    functional = db.Column(db.Boolean(), default=False, server_default="false")
    location = db.Column(db.Integer, db.ForeignKey('locations.id'))
    date_in = db.Column(db.Date, default=dt.date.today())
    date_out = db.Column(db.Date)

    problems = association_proxy('meter_problems', 'problem')

class Problem(db.Model):
    __tablename__ = 'problem'
    id = db.Column(db.Integer, primary_key=True, nullable=False)
    problem = db.Column('keyword', db.String(64))

    def __init__(self, problem):
        self.problem = problem

meterproblems_table = db.Table(
    'meterproblems',
    db.metadata,
    db.Column(
        'meter_id',
        db.Integer,
        db.ForeignKey("meters.id"),
        primary_key=True
    ),
    db.Column(
        'problem_id',
        db.Integer,
        db.ForeignKey("problem.id"),
        primary_key=True
    )
)

你的查询会变成:

meters = Meter.query.filter(
    extract('month', Meter.date_in) == month,
    Meter.meter_problems.has(keyword=prob_id)
).all()

assocation_proxy 用于轻松地将项目添加到 meter_problems 列:

obj.problems.append('something')

其中 obj 是一个仪表对象。

文档 here