如何基于Python Peewee动态构建带过滤条件的查询?
实现动态多过滤条件的User查询
首先修正你User模型中的拼写错误:BooleandField 应该改为 BooleanField,修正后的模型示例(基于Peewee ORM):
from peewee import Model, CharField, BooleanField, SqliteDatabase # 根据实际使用的数据库替换,这里以SQLite为例 db = SqliteDatabase('user_db.db') class User(Model): username = CharField() email = CharField() disabled = BooleanField() admin = BooleanField() # 修正拼写错误 class Meta: database = db
接下来可以直接基于你给出的结构完善代码,实现动态叠加过滤条件的效果:
过滤函数与查询函数实现
def apply_filter(query, filter_name): match filter_name: case "disabled": return query.where(User.disabled == True) case "admin": return query.where(User.admin == True) case "protonmail": # 两种写法都可以,endswith更直观,%是Peewee的模糊匹配语法 return query.where(User.email.endswith('protonmail.com')) # return query.where(User.email % "%protonmail.com") # 可继续扩展更多过滤规则 case _: # 遇到未知过滤条件时,返回原查询(也可根据需求抛出异常) return query def get_users(filters): # 初始化基础查询 query = User.select() # 依次叠加所有过滤条件 for filter_item in filters: query = apply_filter(query, filter_item) # 直接将查询结果转为列表,无需手动循环append return list(query)
使用示例
# 获取同时满足「禁用」「管理员」「ProtonMail邮箱」的用户 target_users = get_users(["disabled", "admin", "protonmail"]) for user in target_users: print(f"用户名: {user.username}, 邮箱: {user.email}")
扩展:支持带参数的复杂过滤
如果需要更灵活的过滤(比如指定任意邮箱域名、用户名前缀),可以把过滤项改为字典格式,传递过滤类型和参数:
def apply_filter(query, filter_spec): filter_type = filter_spec["type"] match filter_type: case "disabled": return query.where(User.disabled == True) case "admin": return query.where(User.admin == True) case "email_domain": domain = filter_spec["value"] return query.where(User.email.endswith(f"@{domain}")) case "username_prefix": prefix = filter_spec["value"] return query.where(User.username.startswith(prefix)) case _: return query # 使用示例 custom_filters = [ {"type": "admin"}, {"type": "email_domain", "value": "gmail.com"} ] gmail_admin_users = get_users(custom_filters)
可选:添加异常处理
如果需要处理数据库连接或查询异常,可以在get_users中补充捕获逻辑:
def get_users(filters): try: query = User.select() for filter_item in filters: query = apply_filter(query, filter_item) return list(query) except Exception as e: # 可根据需求记录日志或返回错误提示 print(f"查询失败: {str(e)}") return []
内容的提问来源于stack exchange,提问作者Kreuterkeule
相关产品推荐
相关产品推荐

