SQLAlchemy 和 Python 中的条件过滤

• 下载 conditional_query.zip - 34.7 KB

引言

在 SQLAlchemy 中，我们熟悉 filter 和 filter_by。但是，如果我们需要进行条件过滤怎么办？这里有一个创建自定义条件过滤运算符的例子。

背景

这是另一种常见的情况，当过滤属性的值为 null/empty 或不满足特定需求时，我们需要忽略 WHERE 条件的一部分。就像下图所示

 

之前我曾在 SQL 和 Entity Framework, C# 中使用过类似的方法。所以我认为同样的方法也可以应用于 SQLAlchemy 和 Python。

查询助手

自定义查询类

为 SQLAlchemy 创建一个自定义查询类，以包含新的 filter_if 方法。

这里如果 condition 为真，则 *criterion 将被应用为过滤器

查看 query_helper.py

from sqlalchemy.orm import Query

class CustomQuery(Query):
    def filter_if(self: Query, condition: bool, *criterion):
        if condition:
            return self.filter(*criterion)
        else:
            return self

使用带有 Db Session 的查询类

将这个自定义查询类添加到 SQLAlchemy 会话中，以便使用新创建的 filter_if 方法。

database.py 

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
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 所有表模型的通用或基类
• GroupQueue DB 表，我们将在其中使用新的运算符

models.py

class AppBaseModelOrm:
    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    is_active = Column(Boolean, default=True)  # soft delete
    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)


class GroupQueue(AppBaseModelOrm, Base):
    __tablename__ = "group_queues"
    name = Column(String, index=True)

条件过滤 

过滤模型

这是一个 FastApi pydantic 模型

schemas.py 

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

使用 filter_if

这里 GroupQueueCrud 是一个 CRUD 助手类，请查看 search 方法，该方法正在调用或使用 filter_if 方法

group_queue_crud.py 

from sqlalchemy.orm import Session
from app import schemas
from db import models
from db.table_repo import TableRepository

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("%{}%".format(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

使用代码

Go to backend folder
Open cmd 
Type docker-compose up -d

\backend> docker-compose up -d

project will run https://:4003

Go to Api Doc
https://:4003/docs#/

参考文献

使用自定义查询类 https://stackoverflow.com/questions/15936111/sqlalchemy-can-you-add-custom-methods-to-the-query-object

历史

• 2022 年 7 月 3 日