Click here to Skip to main content
15,868,016 members
Articles / Database Development

Conditional filtering in SQLAlchemy and Python

Rate me:
Please Sign up or sign in to vote.
1.67/5 (3 votes)
3 Jul 2022CPOL1 min read 15.1K   45   2   2
Conditional filter query example in SQLAlchemy and Python
This is an example to create a custom conditional filter operator.

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

Python
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 

Python
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

Python
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 

Python
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 

Python
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

Python
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

License

This article, along with any associated source code and files, is licensed under The Code Project Open License (CPOL)


Written By
Bangladesh Bangladesh
This member has not yet provided a Biography. Assume it's interesting and varied, and probably something to do with programming.

Comments and Discussions

 
GeneralAnother approach Pin
ymezencev21-Mar-23 8:21
ymezencev21-Mar-23 8:21 
GeneralRe: Another approach Pin
DiponRoy23-Mar-23 7:35
mvaDiponRoy23-Mar-23 7:35 

General General    News News    Suggestion Suggestion    Question Question    Bug Bug    Answer Answer    Joke Joke    Praise Praise    Rant Rant    Admin Admin   

Use Ctrl+Left/Right to switch messages, Ctrl+Up/Down to switch threads, Ctrl+Shift+Left/Right to switch pages.