在FastAPI中实现类似Django-filter的大于、小于等范围筛选功能,可以通过结合Pydantic模型和SQLAlchemy ORM来实现。以下是一个高效的实现方案:
首先,确保你已经安装了FastAPI、SQLAlchemy和Pydantic:
pip install fastapi sqlalchemy pydantic
定义一个Pydantic模型来接收查询参数,这些参数将用于筛选数据。
from pydantic import BaseModel
from typing import Optional
class FilterParams(BaseModel):
min_value: Optional[int] = None
max_value: Optional[int] = None
# 你可以根据需要添加更多的筛选条件
定义一个SQLAlchemy模型来表示数据库中的表。
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
value = Column(Integer)
在FastAPI应用中,使用SQLAlchemy进行数据库操作,并根据查询参数动态构建查询。
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from sqlalchemy import and_
app = FastAPI()
# 假设你已经配置好了数据库连接
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 依赖项,用于获取数据库会话
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/items/")
def get_items(filter_params: FilterParams = Depends(), db: Session = Depends(get_db)):
query = db.query(Item)
if filter_params.min_value is not None:
query = query.filter(Item.value >= filter_params.min_value)
if filter_params.max_value is not None:
query = query.filter(Item.value <= filter_params.max_value)
items = query.all()
return items
运行FastAPI应用,你可以通过以下URL进行测试:
/items/?min_value=10
:获取value
大于等于10的条目。/items/?max_value=20
:获取value
小于等于20的条目。/items/?min_value=10&max_value=20
:获取value
在10到20之间的条目。你可以根据需要扩展FilterParams
模型,添加更多的筛选条件,并在查询中动态应用这些条件。
value
)创建索引,以提高查询性能。完整的示例代码如下:
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from sqlalchemy import create_engine, Column, Integer, String, and_
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from pydantic import BaseModel
from typing import Optional
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
value = Column(Integer)
class FilterParams(BaseModel):
min_value: Optional[int] = None
max_value: Optional[int] = None
app = FastAPI()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/items/")
def get_items(filter_params: FilterParams = Depends(), db: Session = Depends(get_db)):
query = db.query(Item)
if filter_params.min_value is not None:
query = query.filter(Item.value >= filter_params.min_value)
if filter_params.max_value is not None:
query = query.filter(Item.value <= filter_params.max_value)
items = query.all()
return items
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
通过结合Pydantic模型和SQLAlchemy ORM,你可以在FastAPI中高效地实现类似Django-filter的大于、小于等范围筛选功能。这种方法灵活且易于扩展,适用于各种复杂的筛选需求。