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