如何在FastAPI+SQLAlchemy中实现动态字段distance_to的过滤？

你遇到的核心问题是：query_expression定义的distance_to只是ORM实体的一个占位符字段，仅通过with_expression作为查询选项添加时，只会在ORM加载User实例时填充该字段的值，但不会把距离计算逻辑注入到WHERE子句中。fastapi_filter生成过滤条件时，直接引用User.distance_to，但此时这个字段在WHERE阶段没有对应的计算逻辑，所以被解析为NULL，导致过滤失效。

下面提供两种可行的解决方式，你可以根据自己的业务场景选择：

方案1：将计算逻辑整合到查询，基于表达式过滤 #

直接在查询中定义距离计算逻辑，同时用该逻辑做过滤，再通过with_expression把结果映射到User.distance_to字段：

from sqlalchemy import func, select
from sqlalchemy.orm import with_expression

async def get_users(
    session: AsyncSession, user_filter: UserFilter, current_user: User
):
    # 替换成你实际的距离计算逻辑，比如用地理空间函数或自定义公式
    distance_calc = func.ST_Distance(User.location.geom, current_user.location.geom)
    
    query = (
        select(User)
        .join(User.location)
        .options(
            with_expression(User.distance_to, distance_calc)
        )
    )
    
    # 手动处理distance_to的过滤条件
    if user_filter.distance_to__lt is not None:
        query = query.filter(distance_calc < user_filter.distance_to__lt)
    
    # 应用其他过滤器和排序
    query = user_filter.filter(query)
    query = user_filter.sort(query)
    
    result = await session.execute(query)
    return result.scalars().all()

同时修改UserFilter，移除自动生成错误条件的字段：

from fastapi_filter.contrib.sqlalchemy import Filter
from typing import Optional

class UserFilter(Filter):
    # 保留你的其他过滤字段
    order_by: list[str] = ['first_name']

    class Constants(Filter.Constants):
        model = User

方案2：重写过滤器的过滤方法，自定义处理逻辑 #

如果想保留distance_to__lt的过滤器字段，可以重写Filter的filter方法，手动注入计算逻辑：

from fastapi_filter.contrib.sqlalchemy import Filter
from typing import Optional
from sqlalchemy import func

class UserFilter(Filter):
    # 保留你的其他过滤字段
    distance_to__lt: Optional[float] = None
    order_by: list[str] = ['first_name']

    class Constants(Filter.Constants):
        model = User
    
    def filter(self, query, current_user):
        # 先处理其他字段的默认过滤逻辑
        query = super().filter(query)
        
        # 单独处理distance_to的过滤
        if self.distance_to__lt is not None:
            # 替换成你实际的距离计算逻辑
            distance_calc = func.ST_Distance(self.Constants.model.location.geom, current_user.location.geom)
            query = query.filter(distance_calc < self.distance_to__lt)
        
        return query

然后在CRUD代码中调用时传入current_user：

async def get_users(
    session: AsyncSession, user_filter: UserFilter, current_user: User
):
    distance_calc = func.ST_Distance(User.location.geom, current_user.location.geom)
    
    query = (
        select(User)
        .join(User.location)
        .options(
            with_expression(User.distance_to, distance_calc)
        )
    )
    
    # 传入current_user给自定义过滤方法
    query = user_filter.filter(query, current_user)
    query = user_filter.sort(query)
    
    result = await session.execute(query)
    return result.scalars().all()

• query_expression只是一个占位符，不能直接用于WHERE子句过滤，必须基于实际的计算表达式来做条件判断。
• 如果使用地理空间计算（比如PostGIS），要确保数据库支持对应的函数，且SQLAlchemy能正确识别这些函数。

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