在数据库操作中，经常需要根据特定的条件来过滤查询结果。例如，在Web应用中，用户可能会通过表单提交一些过滤条件，希望根据这些条件来动态地构建查询。在Python的SQLAlchemy库中，可以通过定义自定义的查询类和方法来实现这一功能。本文将介绍如何创建一个自定义的条件过滤操作符。
在许多情况下，需要忽略一部分WHERE条件，当过滤属性的值为null/空或者不满足特定需求时。例如，可能需要在SQL查询中使用类似的功能，并且希望在SQLAlchemy和Python中实现相同的功能。
自定义查询类
为了在SQLAlchemy中包含新的filter_if方法，需要创建一个自定义的查询类。
from sqlalchemy.orm import Query
class CustomQuery(Query):
def filter_if(self, condition: bool, *criterion):
if condition:
return self.filter(*criterion)
else:
return self
如果条件为真，则将criterion作为过滤条件应用。
将自定义查询类添加到SQLAlchemy会话中，以便使用新创建的filter_if方法。
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from db.query_helper import CustomQuery
SQLALCHEMY_DATABASE_URL = "sqlite:///repo_app/data/test_sql_app.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, query_cls=CustomQuery)
Base = declarative_base()
这样就可以在数据库会话中使用自定义的查询类了。
AppBaseModelOrm是一个通用的或基类，用于所有表模型。
class AppBaseModelOrm:
id = Column(Integer, primary_key=True, index=True, autoincrement=True)
is_active = Column(Boolean, default=True)
created_by = Column(Integer)
updated_by = Column(Integer, default=None)
created_datetime = Column(DateTime(timezone=True), default=datetime.datetime.utcnow)
updated_datetime = Column(DateTime(timezone=True), default=None, onupdate=datetime.datetime.utcnow)
account_id = Column(Integer)
GroupQueue是一个数据库表，将在其中使用新的操作符。
Filter Model是一个FastApi pydantic模型。
class GroupQueueFilter(CamelModel):
account_ids: List[int] = []
name: Optional[str] = None
is_active: Optional[bool] = None
from_created_datetime: Optional[datetime.datetime] = None
to_created_datetime: Optional[datetime.datetime] = None
在这里，GroupQueueCrud是一个CRUD帮助类，检查search方法，它调用或使用filter_if方法。
class GroupQueueCrud(TableRepository):
def __init__(self, db: Session):
super().__init__(db=db, entity=models.GroupQueue)
def search(self, filter: schemas.GroupQueueFilter):
data = self.db.query(models.GroupQueue) \
.filter_if(filter.account_ids is not None and len(filter.account_ids), models.GroupQueue.account_id.in_(filter.account_ids)) \
.filter_if(filter.is_active is not None, models.GroupQueue.is_active == filter.is_active) \
.filter_if(filter.name is not None, models.GroupQueue.name.ilike(f"%{filter.name}%")) \
.filter_if(filter.from_created_datetime is not None, filter.from_created_datetime and models.GroupQueue.created_datetime >= filter.from_created_datetime) \
.filter_if(filter.to_created_datetime is not None, filter.to_created_datetime and models.GroupQueue.created_datetime <= filter.to_created_datetime)
return data
使用这段代码，可以在数据库查询中根据条件动态地添加过滤条件。
要运行项目，请转到后端文件夹，打开命令行，输入以下命令：
docker-compose up -d
项目将在http://localhost:4003上运行。
要查看API文档，请访问http://localhost:4003/docs。