FastAPI ORM实现BETWEEN查询：优化日期范围查询遗留代码

有两种高效的实现方式，均直接在数据库层面完成日期区间过滤，避免全量查询后再做内存过滤：

方式一：扩展现有get_all方法 #

直接给原方法添加日期区间参数，兼容原有功能的同时支持日期过滤：

from datetime import datetime
from sqlalchemy import select, desc, and_
from sqlalchemy.ext.asyncio import AsyncSession

async def get_all(
            self,
            session: AsyncSession,
            order_field: str | None = 'created_at',
            is_desc: bool | None = False,
            start_date: datetime | None = None,
            end_date: datetime | None = None,
            **param
    ):
        # 处理排序逻辑
        if order_field is not None:
            k = desc(order_field) if is_desc else order_field
        else:
            k = 'created_at'
        
        # 构建基础查询
        query = select(self.__model).filter_by(**param).order_by(k)
        
        # 添加日期区间过滤（假设注册日期字段是`created_at`，请替换为实际字段名）
        if start_date and end_date:
            query = query.filter(
                and_(
                    self.__model.created_at >= start_date,
                    self.__model.created_at <= end_date
                )
            )
        elif start_date:
            query = query.filter(self.__model.created_at >= start_date)
        elif end_date:
            query = query.filter(self.__model.created_at <= end_date)
        
        objs = await session.scalars(query)
        return objs.all()

使用示例：

# 查询2023-01-01到2023-12-31之间注册的对象
await repo.get_all(
    session,
    start_date=datetime(2023,1,1),
    end_date=datetime(2023,12,31,23,59,59)
)

方式二：新增独立的日期区间查询方法 #

如果不想修改原有方法，新增专门方法更符合单一职责原则：

async def get_by_register_date_range(
            self,
            session: AsyncSession,
            start_date: datetime,
            end_date: datetime,
            order_field: str | None = 'created_at',
            is_desc: bool | None = False,
            **param
    ):
        # 处理排序逻辑
        if order_field is not None:
            k = desc(order_field) if is_desc else order_field
        else:
            k = 'created_at'
        
        query = (
            select(self.__model)
            .filter_by(**param)
            .filter(
                and_(
                    self.__model.created_at >= start_date,
                    self.__model.created_at <= end_date
                )
            )
            .order_by(k)
        )
        
        objs = await session.scalars(query)
        return objs.all()

关键说明 #

• 字段替换：如果你的注册日期字段不是created_at，将代码中self.__model.created_at替换为实际字段（如register_date）
• 边界调整：根据业务需求选择>=/<=或>/<，决定是否包含区间端点的时间
• 性能优势：这种方式直接生成带WHERE条件的SQL，数据库可利用字段索引（若已创建），比全查后过滤效率提升显著

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