如何基于SQLAlchemy为Python API实现动态查询过滤器？

前端通过查询字符串传递可选过滤参数，示例请求：

http://mybackend.com/api?first_name=John&last_name=Doe

后端需根据参数动态过滤数据库，无参数时返回所有用户数据。当前尝试的实现方式无法生效，寻求可行思路。

用户尝试的无效实现示例：

• 拼接查询字符串传入filter_by：

def filter_results():
    filter_obj = {
        'first_name' : request.args.get('first_name', null, type=str),
        'last_name' : request.args.get('last_name', null, type=str),
        'city' : request.args.get('city', null, type=str),
        'country' : request.args.get('country', null, type=str),
        'phone' : request.args.get('phone', null, type=int)
    }
    query_string = ''
    for k, v in filter_obj.items():
        if v != null:
            query_string += f'{k} = {v}, '
    query_string.rstrip(', ')
    user_profile = Users.query.filter_by(query_string).all()
    response = users_schema.dump(user_profile)
    return jsonify(response)

• 尝试用text()配合filter_by：

from sqlalchemy import text
filter_string = 'first_name = "John"'
Users.query.filter_by(text(filter_string)).all()

可行实现思路 #

方法1：利用filter_by的关键字参数特性 #

filter_by支持直接传入关键字参数字典，我们只需要构建包含非空参数的字典即可：

from flask import request, jsonify

def filter_results():
    # 定义允许过滤的字段白名单，防止非法参数
    allowed_fields = {'first_name', 'last_name', 'city', 'country', 'phone'}
    # 提取有效的查询参数
    filter_params = {}
    for key, value in request.args.items():
        if key in allowed_fields:
            # 处理phone字段的类型转换
            if key == 'phone':
                try:
                    filter_params[key] = int(value)
                except ValueError:
                    # 忽略无效的phone参数
                    continue
            else:
                filter_params[key] = value
    
    # 动态过滤：无参数时返回所有用户
    user_profiles = Users.query.filter_by(**filter_params).all()
    response = users_schema.dump(user_profiles)
    return jsonify(response)

方法2：使用filter配合模型字段（支持复杂逻辑） #

如果需要模糊查询、范围匹配等复杂过滤，推荐使用filter结合模型字段对象：

from flask import request, jsonify
from sqlalchemy import and_

def filter_results():
    filters = []
    
    # 逐个处理参数，构建过滤条件
    first_name = request.args.get('first_name', type=str)
    if first_name:
        filters.append(Users.first_name == first_name)
    
    last_name = request.args.get('last_name', type=str)
    if last_name:
        filters.append(Users.last_name == last_name)
    
    city = request.args.get('city', type=str)
    if city:
        filters.append(Users.city == city)
    
    country = request.args.get('country', type=str)
    if country:
        filters.append(Users.country == country)
    
    phone = request.args.get('phone', type=int)
    if phone:
        filters.append(Users.phone == phone)
    
    # 组合所有条件，无过滤条件时返回全部数据
    user_profiles = Users.query.filter(and_(*filters)).all()
    response = users_schema.dump(user_profiles)
    return jsonify(response)

无效原因说明 #

• filter_by仅接受关键字参数（如filter_by(first_name='John')），不支持字符串格式的查询语句
• text()需要配合filter使用，而非filter_by，且正确用法需绑定参数（如filter(text("first_name = :name")).params(name='John')），但这种方式不如直接使用模型字段安全

关键注意事项 #

• 字段白名单：必须限制允许过滤的字段，避免恶意参数注入（比如is_admin=true）
• 类型校验：查询参数默认是字符串，需对数字、日期等类型做转换，并处理转换失败的情况
• SQL注入防护：避免直接拼接字符串到查询中，上述两种方法均使用SQLAlchemy的参数绑定机制，天然防护注入

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