Conditional filtering in SQLAlchemy and Python

• Download conditional_query.zip - 34.7 KB

Introduction

In SQLAlchemy we are familiar with filter and filter_by. But what if we need to do conditional filtering. Here is an example of creating a custom conditional filter operator.

Background

This is another common situation, where we need to ignore a part of WHERE conditions when the value of a filter property is null/empty or not fulfilling specific needs. Like the below image:

Previsiaoy I had used similar things with SQL And Entity Framework, C#. So I thought the same could be done with SQLAlchemy and python.

Query Helper

Custom Query Class

Creating a customer query class for SQLAlchemy to include the new filter_if method.

Here is the condition is true, the *criterion will be applied as filter

Check 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 

Use Query Class With Db Session

Adding this custom query class into the SQLAlchemy session, to use the newly created filter_if method.

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()

Table

• AppBaseModelOrm Common or base class for all table models
• GroupQueue DB table, where we will use the new operator

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)

Conditional Filtering 

Filter Model

This is a FastApi pydantic model

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

Using filter_if

Here GroupQueueCrud is a CRUD helper class, check the search method which is calling or using filter_if method

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 

Using the code

 Go to backend folder Open cmd  Type docker-compose up -d \backend> docker-compose up -d project will run http://localhost:4003 Go to Api Doc http://localhost:4003/docs#/ 

References

Use custom query class https://stackoverflow.com/questions/15936111/sqlalchemy-can-you-add-custom-methods-to-the-query-object

History

• 3rd July 2022