实现Python字典列表的query、select、field_filter查询过滤函数

现有如下字典列表：

friends = [
{'name': 'Sam', 'gender': 'male', 'sport': 'Basketball'},
{'name': 'Emily', 'gender': 'female', 'sport': 'volleyball'},
]

需要实现query、select、field_filter三个函数，用于处理此类字典列表，支持选择指定列并按列进行过滤。

示例用法 #

result = query(friends, 
select('name', 'gender', 'sport'), 
field_filter('sport', *('Basketball', 'volleyball')), 
field_filter('gender', *('male',)))

# 输出结果
[{'gender': 'male', 'name': 'Sam', 'sport': 'Basketball'}]

预定义代码框架 #

from typing import Dict, Any, Callable, Iterable
DataType = Iterable[Dict[str, Any]]
ModifierFunc = Callable[[DataType], DataType]
def query(data: DataType, selector: ModifierFunc,
      *filters: ModifierFunc) -> DataType:
"""Query data with column selection and filters
:param data: List of dictionaries with columns and values
:param selector: result of `select` function call
:param filters: Any number of results of `field_filter` function calls
:return: Filtered data"""
pass
def select(*columns: str) -> ModifierFunc:
"""Return function that selects only specific columns from dataset"""
pass
def field_filter(column: str, *values: Any) -> ModifierFunc:
"""Return function that filters specific column to be one of `values`"""
pass

完整实现代码 #

from typing import Dict, Any, Callable, Iterable

DataType = Iterable[Dict[str, Any]]
ModifierFunc = Callable[[DataType], DataType]


def query(data: DataType, selector: ModifierFunc,
          *filters: ModifierFunc) -> DataType:
    """Query data with column selection and filters
    :param data: 包含列和值的字典列表
    :param selector: `select`函数的返回结果
    :param filters: 任意数量的`field_filter`函数返回结果
    :return: 过滤后的数据"""
    filtered_data = data
    # 依次应用所有过滤条件
    for filter_func in filters:
        filtered_data = filter_func(filtered_data)
    # 最后执行列选择
    return selector(filtered_data)


def select(*columns: str) -> ModifierFunc:
    """返回一个仅从数据集中选择指定列的函数"""
    def selector(data: DataType) -> DataType:
        # 仅保留每个字典中指定的列
        return [{col: item[col] for col in columns if col in item} for item in data]
    return selector


def field_filter(column: str, *values: Any) -> ModifierFunc:
    """返回一个过滤函数，保留指定列值在给定范围内的项"""
    def filter_func(data: DataType) -> DataType:
        # 筛选出指定列值属于目标集合的字典
        return [item for item in data if item.get(column) in values]
    return filter_func

实现说明 #

• field_filter：生成的过滤函数会检查每个字典的指定列值是否在传入的values中，用item.get(column)避免因字典缺少该列而抛出KeyError。
• select：生成的选择函数会遍历每个字典，仅保留传入的columns列表中存在的键值对，确保兼容性。
• query：先按顺序应用所有过滤条件对数据进行筛选，最后应用列选择得到最终结果，保证过滤逻辑优先于列选择执行。

内容的提问来源于stack exchange，提问作者Yuliya Yeusiyevich