FastAPI中如何高效实现类似Django filter的范围查询？
最新回答
瑾沫流年
2020-07-14 18:36:48
在FastAPI中高效实现类似Django filter的范围查询，可通过Pydantic模型定义查询参数 + SQLAlchemy动态构建查询条件实现。以下是完整实现方案：
1. 定义Pydantic查询参数模型使用BaseModel定义前端可传递的查询参数，支持范围查询（如最小值/最大值、日期区间等）：
from pydantic import BaseModelfrom typing import Optional, Listfrom datetime import dateclass SeriesFilter(BaseModel): company: Optional[List[int]] = None # 多值筛选（IN查询） confirm_start: Optional[date] = None # 日期范围下限 confirm_end: Optional[date] = None # 日期范围上限 score_min: Optional[int] = None # 数值范围下限 score_max: Optional[int] = None # 数值范围上限2. 配置SQLAlchemy数据库连接在database.py中定义数据库模型和会话管理：
from sqlalchemy import create_engine, Column, Integer, Datefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmakerBase = declarative_base()class Series(Base): __tablename__ = "series" id = Column(Integer, primary_key=True) company_id = Column(Integer) confirm_at = Column(Date) rating_score = Column(Integer)# 数据库连接配置（需替换为实际URL）SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"engine = create_engine(SQLALCHEMY_DATABASE_URL)SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)def get_db(): db = SessionLocal() try: yield db finally: db.close()3. 实现动态查询路由在FastAPI路由中，通过依赖注入获取数据库会话，并根据参数动态构建查询：
from fastapi import FastAPI, Dependsfrom sqlalchemy.orm import Sessionfrom datetime import dateapp = FastAPI()@app.get("/series/")async def read_series( db: Session = Depends(get_db), filter_params: SeriesFilter = Depends() # 直接依赖Pydantic模型): query = db.query(Series) # 动态构建查询条件 if filter_params.company: query = query.filter(Series.company_id.in_(filter_params.company)) if filter_params.confirm_start: query = query.filter(Series.confirm_at >= filter_params.confirm_start) if filter_params.confirm_end: query = query.filter(Series.confirm_at <= filter_params.confirm_end) if filter_params.score_min is not None: # 显式检查None query = query.filter(Series.rating_score >= filter_params.score_min) if filter_params.score_max is not None: query = query.filter(Series.rating_score <= filter_params.score_max) return query.all()4. 关键实现细节- 动态条件构建：通过if语句逐个检查参数，仅对非空参数添加过滤条件。
- 范围查询支持：
日期范围：confirm_start和confirm_end分别对应>=和<=。
数值范围：score_min和score_max实现最小/最大值筛选。
多值筛选：company列表通过in_()实现IN查询。
- 依赖注入：get_db()通过生成器管理会话生命周期，避免连接泄漏。
- SQLModel集成：若使用SQLModel（基于SQLAlchemy的简化版），可进一步简化模型定义：from sqlmodel import SQLModel, Fieldclass Series(SQLModel, table=True): id: Optional[int] = Field(primary_key=True) company_id: int confirm_at: date rating_score: int
- 查询参数验证：在Pydantic模型中添加参数验证逻辑，例如：from pydantic import validatorclass SeriesFilter(BaseModel): score_min: Optional[int] = None score_max: Optional[int] = None @validator('score_max') def check_score_range(cls, v, values): if 'score_min' in values and v < values['score_min']: raise ValueError('score_max must be >= score_min') return v
前端可通过以下URL触发查询：
GET /series/?company=1&company=2&confirm_start=2023-01-01&score_min=80返回结果为符合所有条件的记录列表。
此方案通过模型定义参数、动态条件拼接和依赖注入管理会话，实现了类似Django filter的灵活查询能力，同时保持了FastAPI的轻量级特性。
热门标签
