Sqlalchemy：使用兩個ARRAY類型列的過濾操作

Question

我有兩個類：Question和User。 Question類有一個類型的字段，其類型爲Column(ARRAY(Integer))，而User類有一個對象preferences，該對象的字段名爲ignored_categories，也是Column(ARRAY(Integer))類型。

我想查詢所有沒有在出現在ignored_categories字段從用戶的喜好或者以前是真的category_ids場至少一個值的問題，必須在category_ids領域的任何其他值不在用戶的偏好字段ignore_categories中。因此，下面列出的是真實的：

• category_ids = []和ignored_categories = [1，2，3] =>應該通過。
• category_ids = [1,2,4]和ignored_categories = [1,2,3] =>應該通過。
• category_ids = [1]和ignored_categories = [1，2，3] =>應該忽略。
• category_ids = [1，2，3]和ignored_categories = [1,2,3] =>應該忽略。
• category_ids = [1,2,3,4]和ignored_categories = [1,2,3] =>應該通過。
• category_ids = [5,7]和ignored_categories = [1,2,3] =>應該通過。

這是我想出了這麼遠：

user = DBSession.query(User).filter_by(id=user_id).one() # this gets the user object 
query = DBSession.query(Question).order_by(Question.created_at.desc()) 
query = query.filter(
    or_(
     not_(
      Question.category_ids.overlap(user.preferences.ignored_categories) 
     ), 
     Question.category_ids.contains(user.preferences.ignored_categories) 
    ) 
) 

這樣做的問題是，contains只有測試，如果category_ids是ignored_categories一個超集，這不能給出數據的正確結果集，如：

• category_ids = [1]和ignored_categories = [1，2，3] =>應忽略。

因此，它需要另一個條件來測試是否至少有一個值匹配，而其他條件不匹配。

的overlap功能僅當存在於兩個陣列的至少一個值的測試，但是此操作失敗測試，​​如：

• category_ids = [1，2，4]和ignored_categories = [1，2 ，3] =>應該通過。

哪個應該通過，但它沒有，因爲我需要否定操作。如果我不否定它，它只會過濾沒有任何共同點的數組。

編輯：這是我的表看起來像：

class User(Base, DictSerializable): 
    __tablename__ = 'users' 
    __table_args__ = dict(schema='user') 

    id = Column(types.Id, primary_key=True) 
    # other fields 


class UserPreferences(Base, DictSerializable): 
    __tablename__ = 'user_preferences' 
    __table_args__ = dict(schema='user') 

    id = Column(types.Id, primary_key=True) 
    user_id = Column(types.Id, ForeignKey(User.id)) 
    ignored_categories = Column(types.ARRAY(types.Number), default=[]) 
    # other fields 

    user = relationship("User", 
         backref=backref("preferences", single_parent=True, cascade="all, delete-orphan", 
             passive_deletes=True, uselist=False), 
         ) 

class Question(Base, DictSerializable): 
    __tablename__ = 'questions' 
    __table_args__ = dict(schema='question') 

    id = Column(types.Id, primary_key=True) 
    user_id = Column(types.Id, ForeignKey(User.id, ondelete="CASCADE", onupdate="CASCADE")) 

    # other fields 

    category_ids = Column(types.ARRAY(types.Integer)) 

    user = relationship("User", foreign_keys=user_id, 
         backref=backref("questions", order_by=id, single_parent=True, uselist=True, 
             cascade="all, delete-orphan", passive_deletes=True) 
         ) 


question_categories = Table('question_categories', Base.metadata, 
    Column('question_id', types.Integer, ForeignKey(Question.id)), 
    Column('category_id', types.Integer, ForeignKey(Category.id))) 
Question.categories = relationship(Category, secondary=question_categories, backref=backref('questions')) 

Answer 1

這個問題相當於在category_ids找到一個元素不ignored_categories存在，與在那裏如果category_ids爲空，則不能忽略的特殊情況。

這可以實現連接表如下：

SELECT DISTINCT questions.* 
FROM questions 
LEFT JOIN question_categories ON questions.id = question_categories.question_id 
WHERE (
    question_categories.category_id IS NULL -- special case where question has no categories 
    OR question_categories.category_id NOT IN (
    SELECT category_id FROM user_ignored_categories WHERE user_id = 123 -- for this user 
) -- find a category that isn't ignored 
) 

要使用數組做到這一點，你可以這樣做：

SELECT questions.* 
FROM questions 
WHERE (
    cardinality(categories) = 0 -- special case where question has no categories 
    OR NOT ((SELECT ignored_categories FROM users WHERE id = 123) @> categories) 
); 

您可以測試這些對你的數據here。