sqlalchemy從一個表中的一列中的多行查詢單個查詢

Question

我遇到了一些與sqlite3和sqlalchemy有關的問題。從某個時候我嘗試做出一些具體的查詢，並以某種方式失敗。數據庫由兩個表用戶和屬性組成。這些表的模式如下所示。

sqlite> .schema users 
CREATE TABLE users (
    id INTEGER NOT NULL, 
    name VARCHAR(50) NOT NULL, 
    PRIMARY KEY (id) 
); 

sqlite> .schema properties 
CREATE TABLE properties (
    id INTEGER NOT NULL, 
    property_number INTEGER, 
    user_id INTEGER, 
    PRIMARY KEY (id), 
    FOREIGN KEY(user_id) REFERENCES users (id) 
); 

用戶表的內容很簡單，但屬性值得一些解釋。在property_number列中，我存儲了不同的屬性，每個屬性都有其唯一的編號，例如：屬性禿頭具有編號3，屬性tan具有編號4等。如果用戶具有多個屬性，則每個屬性在屬性表中佔據一行。我選擇了這種風格，可以輕鬆添加新的屬性，而不會影響遷移和類似的東西。

問題是，不知道如何使查詢包含多個屬性。我目前最好的解決方案是，在單獨的查詢中詢問每一個屬性。這給出了一套列表，兩個不同的列表。一個是給定屬性的正面和一個負面實例（正面是我希望用戶擁有的東西，負面是我不喜歡用戶擁有的東西）。在下一步中，我對這兩個子集進行了區分，並獲得最終列表，其中包含對我的屬性感興趣的用戶ID。然後我查詢這些用戶的名字。這似乎是非常複雜的，也許是，但肯定是醜陋的。我也不喜歡對每一個屬性進行單一查詢。 Python代碼，如果somone有興趣。

def prop_dicts(): 
    """Create dictionaries of properties 
    contained in table properties in db. 

    Returns: 
     touple: 
      prop_names (dict) 
      prom_values (dict).""" 

    prop_names = {'higins': 10000, 
        'tall': 1, 
        'fat': 2, 
        'bald': 3, 
        'tan': 4, 
        'hairry': 5} 
    prop_values = {1000: 'higins', 
        1: 'tal', 
        2: 'fat', 
        3: 'bald', 
        4: 'tan', 
        5: 'hairry'} 
    dictionaries = (prop_names, prop_values) 
    return dictionaries 


def list_of_sets_intersection(set_list): 
    """Makes intersection of all sets in list. 

    Args: 
     param1 (list): list containing sets to check. 

    Returns: 
     set (values): contains intersectred values.""" 

    if not set_list: 
     return set() 
    result = set_list[0] 
    for s in set_list[1:]: 
     result &= s 
    return result 


def list_of_sets_union(set_list): 
    """Makes union of elements in all sets in list. 

    Args: 
     param1 (list): list containing sets to check. 

    Returns: 
     set (values): contains union values.""" 

    if not set_list: 
     return set() 
    result = set_list[0] 
    for s in set_list[1:]: 
     result |= s 
    return result 


def db_search(): 
    """Search database against positiv and negative values. 

    Returns: 
     list (sets): one set in list for every property in 
        table properties db.""" 

    n, v = prop_dicts() 

    positive = [2, 3] 
    negative = [4, 5] 
    results_p = [] 
    results_n = [] 

    #Positive properties. 
    for element in xrange(0, len(positive)): 
     subresult = [] 

     for u_id, in db.query(Property.user_id).\ 
           filter_by(property_number = positive[element]): 
      subresult.append(u_id) 

     subresult = set(subresult) 
     results_p.append(subresult) 

    #Negative properties. 
    for element in xrange(0, len(negative)): 
     subresult = [] 

     for u_id, in db.query(Property.user_id).\ 
           filter_by(property_number = negative[element]): 
      subresult.append(u_id) 

     subresult = set(subresult) 
     results_n.append(subresult) 

    print 'positive --> ', results_p 
    print 'negative --> ', results_n 

    results_p = list_of_sets_intersection(results_p) 
    results_n = list_of_sets_union(results_n) 

    print 'positive --> ', results_p 
    print 'negative --> ', results_n 

    final_result = results_p.difference(results_n) 
    return list(final_result) 


print db_search()  

這是一種在單個查詢中做到這一點的方法嗎？我是數據庫領域的新成員，如果問題的質量似乎不好，我很抱歉。有太多的可能性，我真的不知道如何以「正確」的方式做到這一點。我已經在這個主題上搜索了大量的互聯網，我發現最好的解決方案是包含「WHERE」原因和「AND」運算符。但是如果你將兩個表中的相同列連接起來，那麼這兩個不起作用。

SELECT user_id FROM properties WHERE property_number=3 AND property_number=4; 

或在sqlalchemy。

db.query(User.user_id).join(Property).filter(and_(property_number=3, property_number=4)).all() 

這SQLAlchemy的例子可能包含了一些錯誤，因爲我沒有預覽它，但可以肯定你會明白這個是什麼的地步。

Answer 1

您可以通過使用聚合

SELECT user_id 
FROM properties 
WHERE property_number in (3, 4) 
GROUP BY user_id 
HAVING count(*) = 2 

做到這一點在SQLAlchemy中

from sqlalchemy import func 

properties = [3, 4] 
db.session.query(Property.user_id)\ 
    .filter(Property.property_number.in_(properties))\ 
    .group_by(Property.user_id)\ 
    .having(func.count()==len(properties))\ 
    .all() 

更新

positive = [2, 3] 
negative = [4, 5] 

positive_query = db.session.query(Property.user_id)\ 
    .filter(Property.property_number.in_(positive))\ 
    .group_by(Property.user_id)\ 
    .having(func.count()==len(positive)) 

negative_query = db.session.query(Property.user_id)\ 
    .filter(Property.property_number.in_(negative))\ 
    .distinct() 

final_result = positive_query.except_(negative_query).all()