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 列存储一个 Python 列表，其中包含问题对象的 id

然后我有这个函数尝试查询仪表数据库表并仅获取列表存储中具有特定问题 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 是一个仪表对象。

Doumentation这里