FastAPI查询参数转MongoDB聚合：优化$match阶段实现方案问询

我之前用Flask开发了一个展示自己演奏吉他曲的Web应用，现在正把所有MongoDB聚合逻辑从Flask迁移到FastAPI。其中一项功能支持按category（分类）和/或proficiency（熟练度）过滤歌曲，在Flask中查询字符串会被解析为字典，添加songs.前缀后传入聚合的$match阶段。

我已经通过多条件分支在FastAPI中实现了该功能，但代码过于冗余，希望找到更优雅的$match阶段构建方式，例如：

• 统计参数数量后将其作为字典注入聚合管道
• 在聚合管道中通过循环构建$match条件

查阅FastAPI参数文档后，仅覆盖了参数定义与验证部分，未找到相关实现方法。以下是当前的FastAPI路由代码：

@api.get("/songsets/{username}/songs", response_description="List all songs for users")
async def list_songs(username, category: str | None = None,  proficiency: int | None = None):

    if category or proficiency:
        if category and not proficiency:
            songs_pipeline =[
                { "$match" : { "user_name" : username } },
                { "$unwind" : "$songs"},
                    { "$match" :
                                {"songs.category" : category }
                        },
                { "$project" : { "_id" : 0,
                    "song" : "$songs.song",
                    "artist" : "$songs.artist",
                    }
                    }
                ]
        elif proficiency and not category:
            songs_pipeline =[
                { "$match" : { "user_name" : username } },
                { "$unwind" : "$songs"},
                    { "$match" :[
                            {"songs.category" : category },
                            {"songs.proficiency" : proficiency }
                            ]
                        },
                { "$project" : { "_id" : 0,
                    "song" : "$songs.song",
                    "artist" : "$songs.artist"
                    }
                    }
                ]
        elif proficiency and category:
            songs_pipeline =[
                { "$match" : { "user_name" : username } },
                { "$unwind" : "$songs"},
                    { "$match" :
                                {"songs.proficiency" : proficiency }
                        },
                { "$project" : { "_id" : 0,
                    "song" : "$songs.song",
                    "artist" : "$songs.artist",
                    }
                    }
                ]
        
        songs = await db["userSongSet_dev"].aggregate(songs_pipeline).to_list(100)
        return songs

核心思路是动态构建$match条件字典，彻底消除冗余的多分支判断，同时保持逻辑清晰。

具体实现步骤 #

1. 先定义固定的聚合管道结构，包含匹配用户名、展开数组、字段投影这些不变的阶段
2. 动态收集非空的过滤参数，为每个参数添加songs.前缀后存入条件字典
3. 如果存在过滤条件，将生成的$match阶段插入到管道的对应位置
4. 统一执行聚合查询

优化后的代码 #

@api.get("/songsets/{username}/songs", response_description="List all songs for users")
async def list_songs(username, category: str | None = None, proficiency: int | None = None):
    # 初始化固定结构的聚合管道
    songs_pipeline = [
        {"$match": {"user_name": username}},
        {"$unwind": "$songs"},
        {"$project": {
            "_id": 0,
            "song": "$songs.song",
            "artist": "$songs.artist"
        }}
    ]

    # 动态构建过滤条件字典
    filter_conditions = {}
    if category:
        filter_conditions["songs.category"] = category
    if proficiency:
        filter_conditions["songs.proficiency"] = proficiency

    # 有过滤条件时，插入到$unwind之后、$project之前
    if filter_conditions:
        songs_pipeline.insert(2, {"$match": filter_conditions})

    # 执行聚合查询并返回结果
    songs = await db["userSongSet_dev"].aggregate(songs_pipeline).to_list(100)
    return songs

优化优势 #

• 完全移除冗余分支，新增过滤参数时只需添加一行判断逻辑即可
• 固定逻辑与动态逻辑分离，管道结构更清晰易懂
• 和Flask中的实现思路对齐，保持了跨框架的逻辑一致性

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