SqlAlchemy中or_函数报错：预期2个参数却传入3个求助

问题描述 #

使用Python+SqlAlchemy开发接口时，当请求体传入3个及以上参数，会抛出错误：
POST Error: or_ expected 2 arguments, got 3
仅传入1或2个参数时功能正常，无法定位错误原因，请求协助排查。

相关代码 #

主函数read_uptime #

def read_uptime(logid, filteredData, dateStart, dateEnd, timeStart, timeEnd, elementsForPage, currentPage, filterUptime):
    log.info(f"{logid} read_uptime: Started")

    try:
        # 检查时间范围合法性
        startDateTime, endDateTime = _checkDataInput(timeStart, timeEnd, dateStart, dateEnd)

        # 创建过滤条件
        filters = _createFilter(filteredData, startDateTime, endDateTime, filterUptime)

        # 执行查询
        dataFiltered = uptime_model_db.query.with_entities(
            uptime_model_db.projectId.label('projectId'),
            uptime_model_db.url.label('url'),
            uptime_model_db.timeStamp.label('timeStamp'),
            uptime_model_db.uptime.label('uptime'),
            uptime_model_db.latency.label('latency')
        ).filter(*filters).paginate(per_page=int(elementsForPage + 1), page=int(currentPage), error_out=True)

        # 检查是否有下一页
        nextPage = {
            "currentPage": currentPage,
            "totalElements": len(dataFiltered.items)
        }

        if len(dataFiltered.items) > elementsForPage:
            nextPage["nextPage"] = True
        else:
            nextPage["nextPage"] = False

        # 格式化并返回JSON
        return _createJson(dataFiltered.items, nextPage)

    except Exception as e:
        log.error(f"{logid} read_uptime: function read_uptime returned {e}")
        raise e

报错相关辅助函数filterAppend #

def filterAppend(arrayFilter, urls, projectId, arrayUrl):
    if(len(urls) == 1):
        arrayFilter.append(and_(uptime_model_db.projectId == projectId, uptime_model_db.url == urls[0]))

    if(len(urls) > 1):
        for url in urls:
            arrayUrl.append(uptime_model_db.url == url)
        arrayFilter.append(and_(uptime_model_db.projectId == projectId, or_(*arrayUrl)))

过滤条件生成函数_createFilter #

def _createFilter(filteredData, startDateTime, endDateTime, filterUptime):

    filters = []

    if filteredData is not None:
        arrayFilter = []

        for data in filteredData:
            projectId = data["projectId"]
            urls = data["uptimeUrls"]
            arrayUrl = []
            
            if (len(filteredData) == 1):
                filterAppend(filters, urls, projectId, arrayUrl)

            else:
                filterAppend(arrayFilter, urls, projectId, arrayUrl)

        if(len(filteredData) > 1 or len(arrayFilter) > 1):
            filters.append(or_(*arrayFilter))

    if startDateTime is not None:
        filters.append(str(startDateTime) <= uptime_model_db.timeStamp)

    if startDateTime is not None:
        filters.append(str(endDateTime) >= uptime_model_db.timeStamp)

    if filterUptime == "True":
        filters.append(uptime_model_db.uptime < 100)

    return filters

错误定位 #

报错出现在以下两处代码：

1. arrayFilter.append(and_(uptime_model_db.projectId == projectId, or_(*arrayUrl)))
2. filters.append(or_(*arrayFilter))

原因分析 #

1. or_函数参数限制：旧版本SqlAlchemy的or_（以及and_）函数要求至少传入2个条件参数。当通过*解包列表时，如果列表只有1个元素，就会触发expected 2 arguments的错误。
   • 例如：当arrayFilter中只有1个过滤条件时，执行or_(*arrayFilter)相当于只传了1个参数，不符合函数要求。
2. filterAppend条件判断不严谨：使用两个独立的if判断而非if-elif，可能导致逻辑冗余（虽然当前业务场景下不会触发重复添加，但写法存在隐患）。
3. _createFilter存在笔误：第二个时间条件错误使用startDateTime is not None作为判断条件，导致endDateTime的过滤条件永远不会生效（除非startDateTime存在）。

修复方案 #

1. 修正filterAppend的条件判断 #

将第二个if改为elif，避免逻辑冗余：

def filterAppend(arrayFilter, urls, projectId, arrayUrl):
    if len(urls) == 1:
        arrayFilter.append(and_(uptime_model_db.projectId == projectId, uptime_model_db.url == urls[0]))
    elif len(urls) > 1:
        for url in urls:
            arrayUrl.append(uptime_model_db.url == url)
        arrayFilter.append(and_(uptime_model_db.projectId == projectId, or_(*arrayUrl)))

2. 处理or_单参数场景 #

在_createFilter中，根据arrayFilter的长度决定是否使用or_包裹：

if filteredData is not None:
    arrayFilter = []

    for data in filteredData:
        projectId = data["projectId"]
        urls = data["uptimeUrls"]
        arrayUrl = []
        
        if len(filteredData) == 1:
            filterAppend(filters, urls, projectId, arrayUrl)
        else:
            filterAppend(arrayFilter, urls, projectId, arrayUrl)

    # 修改判断逻辑，处理单参数情况
    if len(arrayFilter) > 0:
        if len(arrayFilter) == 1:
            # 只有一个条件时，直接加入filters，无需or_包裹
            filters.append(arrayFilter[0])
        else:
            filters.append(or_(*arrayFilter))

3. 修正时间条件的判断笔误 #

将endDateTime的判断条件改为正确的变量：

if startDateTime is not None:
    filters.append(str(startDateTime) <= uptime_model_db.timeStamp)

# 修正为判断endDateTime是否存在
if endDateTime is not None:
    filters.append(str(endDateTime) >= uptime_model_db.timeStamp)

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