基于Python Ariadne+GraphQL实现MySQL多列过滤功能方案咨询

1. 定义GraphQL过滤输入类型 #

在Schema中定义通用的过滤输入结构，支持常见比较操作，避免为每个字段单独编写逻辑。以users表（含id/name/age字段）为例：

input UserFilterInput {
  id: IntFilterInput
  name: StringFilterInput
  age: IntFilterInput
}

input IntFilterInput {
  eq: Int
  gt: Int
  lt: Int
  gte: Int
  lte: Int
}

input StringFilterInput {
  eq: String
  contains: String
  startsWith: String
  endsWith: String
}

type Query {
  users(filter: UserFilterInput): [User!]!
}

type User {
  id: Int!
  name: String!
  age: Int!
}

2. 编写通用过滤解析工具函数 #

实现一个工具函数，将GraphQL过滤输入动态转换为MySQL的WHERE子句和参数，避免重复编写字段逻辑：

def build_where_clause(filter_input, table_alias=""):
    where_conditions = []
    params = {}
    if not filter_input:
        return "", params

    for field, filters in filter_input.items():
        column = f"{table_alias}.{field}" if table_alias else field
        for op, value in filters.items():
            param_key = f"{field}_{op}"
            # 根据操作符生成对应SQL条件
            if op == "eq":
                where_conditions.append(f"{column} = :{param_key}")
            elif op == "gt":
                where_conditions.append(f"{column} > :{param_key}")
            elif op == "lt":
                where_conditions.append(f"{column} < :{param_key}")
            elif op == "gte":
                where_conditions.append(f"{column} >= :{param_key}")
            elif op == "lte":
                where_conditions.append(f"{column} <= :{param_key}")
            elif op == "contains":
                where_conditions.append(f"{column} LIKE :{param_key}")
                value = f"%{value}%"
            elif op == "startsWith":
                where_conditions.append(f"{column} LIKE :{param_key}")
                value = f"{value}%"
            elif op == "endsWith":
                where_conditions.append(f"{column} LIKE :{param_key}")
                value = f"%{value}"
            params[param_key] = value

    where_clause = "WHERE " + " AND ".join(where_conditions) if where_conditions else ""
    return where_clause, params

3. 修改现有Resolver集成过滤逻辑 #

假设你的现有Resolver如下：

from ariadne import QueryType
import mysql.connector

query = QueryType()

@query.field("users")
def resolve_users(*_):
    conn = mysql.connector.connect(
        host="localhost",
        user="your_user",
        password="your_password",
        database="your_db"
    )
    cursor = conn.cursor(dictionary=True)
    cursor.execute("SELECT id, name, age FROM users")
    users = cursor.fetchall()
    cursor.close()
    conn.close()
    return users

只需小幅修改，集成通用过滤逻辑：

@query.field("users")
def resolve_users(_, info, filter=None):
    conn = mysql.connector.connect(
        host="localhost",
        user="your_user",
        password="your_password",
        database="your_db"
    )
    cursor = conn.cursor(dictionary=True)
    
    # 生成WHERE子句与参数
    where_clause, params = build_where_clause(filter)
    query = f"SELECT id, name, age FROM users {where_clause}"
    
    # 带参数执行查询，避免SQL注入
    cursor.execute(query, params)
    users = cursor.fetchall()
    
    cursor.close()
    conn.close()
    return users

4. 绑定Schema与Resolver #

确保Schema正确绑定QueryType：

from ariadne import make_executable_schema, load_schema_from_path

type_defs = load_schema_from_path("schema.graphql")
schema = make_executable_schema(type_defs, query)

是否需要修改现有代码？ #

需要，但修改量极小：

1. 在GraphQL Schema中新增过滤输入类型，并修改users查询的参数定义。
2. 添加通用的build_where_clause工具函数。
3. 修改现有resolve_users函数，替换固定SQL查询为动态生成的带过滤逻辑的查询。

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