I am currently refining a common data interfacing concept.
I use filter masks in the form of ['key', 'comparison method', 'value']
for data filtering tasks.
Comparison methods are taken from
CMD = {
"==": lambda x, y: x == y,
"!=": lambda x, y: x != y,
"has": lambda x, y: y in x,
"not_has": lambda x, y: y not in x,
"in": lambda x, y: x in y,
"not_in": lambda x, y: x not in y
}
Now filtering looks like this
filters = [['id', '==', 6], ["type", "!=", "game"]]
with self.session_factory() as session:
result = session.query(my_dataclass).filter(
CMD[f[1]](getattr(my_dataclass, f[0]), f[2]) for f in filters
).first()
Note that my_dataclass
is a dynamically generated mapped class, thus the getattr-approach extracts the attribute from the mapped class (e.g. Media.id
) and forwards it to the lambda expression for comparison resulting in filter(Media.id == 6, Media.type != 'game')
for the example above. This might look absolutely over-complicated but makes sense in the broader view of the common data interface, so please bear with me here.
For batch-operations the concept works like this:
from sqlalchemy import and_, or_, not_
filters = [[['id', '==', 6], ["type", "==", "game"]], [['id', '==', 1], ["type", "!=", "game"]]]
with self.session_factory() as session:
result = session.query(my_dataclass).filter(
or_(
and_(
CMD[f[1]](getattr(my_dataclass, f[0]), f[2]) for f in filter_mask
) for filter_mask in filters)
).all()
The main problem
The concept works just fine for the examples above, until I filter for with !=
on acolumns, that holds a NULL/None value.
When I add ["inactive", "!=", "X"]
to a filter (where 'inactive' is a sqlalchemy.CHAR column that initiates with NULL/None and is set to 'X' in case of soft deletes) the query results in None.
Filtering for ["inactive", "==", None]
works just fine. Filtering with "!="
on not-Null/None fields works just fine as well...
I tried to narrow down the problem's possible causes for some time now but just hit a dead end here. Has anyone stumbled across a similar problem?
Thank you for your time in advance!
Further Information:
Python==3.8, SQLAlchemy==1.4.31
I also tried a more SQLAlchemy-near step-by-step approach with iterative query filtering to sort out the problem:
def filter_query(self, my_dataclass: Any, query: Any, filters: list) -> Any:
"""
Method for filtering query.
:param my_dataclass: Data class.
:param query: Query to filter.
:param filters: Filter masks.
:return: Filtered query.
"""
for filter_mask in filters:
if filter_mask[1] == "==":
query = query.filter(getattr(my_dataclass, filter_mask[0]) == filter_mask[2])
elif filter_mask[1] == "!=":
query = query.filter(not_(getattr(my_dataclass, filter_mask[0]) == filter_mask[2]))
elif filter_mask[1] == "in":
query = query.filter(getattr(my_dataclass, filter_mask[0]).in_(filter_mask[2]))
elif filter_mask[1] == "not_in":
query = query.filter(not_(getattr(my_dataclass, filter_mask[0]).in_(filter_mask[2])))
elif filter_mask[1] == "has":
query = query.filter(getattr(my_dataclass, filter_mask[0]).contains(filter_mask[2]))
elif filter_mask[1] == "not_has":
query = query.filter(not_(getattr(my_dataclass, filter_mask[0]).contains(filter_mask[2])))
return query
But the results are the same.