python - 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)
);

users 表的内容非常简单，但是 properties 值得做一些解释。在 property_number 列中，我存储不同的属性，每个属性都有其唯一的编号，例如:属性 bald 的编号为 3，属性 tan 的编号为 4 等。如果用户有多个属性，则每个属性在属性表中占据一行。我选择这种风格是为了轻松地添加新属性，而不会弄乱迁移之类的事情。

问题是，不知道如何进行包含多个属性的查询。我目前最好的解决方案是，在单独的查询中询问每个属性。这给出了 mi 集合列表，两个不同的集合。一个用于给定属性的正面实例，一个用于负面实例(正面等于我希望用户拥有的东西，负面等于我不希望用户拥有的东西)。在下一步中，我将这两个子集区分开来，并获得最终列表，其中包含对我感兴趣的属性的用户 ID。然后我查询这些用户的姓名。它看起来很复杂，也许是，但肯定是丑陋的。我也不喜欢对每个属性进行单一查询。 Python 代码，如果有人感兴趣的话。

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