How can I apply a SqlAlchemy `BinaryExpression` as a filter on a list?

Question

I have a list of objects of type MyObject,

@dataclasses.dataclass
Class MyObject:
    attr: int

obj_list: List[MyObject] = [ MyObject(23), ... ]

My database model in SqlAlchemy is defined as

class MyObjectModel(Base):
    __tablename__ = "objects"

    attr = Column(Integer, nullable=False)

Now for a sqlalchemy.sql.elements.BinaryExpression such as

expr = MyObjectModel.attr == 25

I would like to filter obj_list using this BinaryExpression without using SqlAlchemy session.

To clarify, I would like to have a function convert such that

filtered_obj_list = [o for o in obj_list if convert(expr, o)]

where expr can be any BinaryExpression.

We can use SqlAlchemy's internal methods too.

P.S. The reason I want to do this is that I want to make my own customized Mock for my database and need this level of specificity.

Peter · Accepted Answer · 2022-10-06 15:53:44Z

This is a very basic solution that will work with the built in python operators.

expr = MyObjectModel.attr == 25

def convert(expr, obj):
    bound_value = expr.right.value
    dataclass_value = getattr(obj, expr.left.key)
    return expr.operator(dataclass_value, bound_value)

filtered_obj_list = [o for o in obj_list if convert(expr, o)]

Edit: To expand some functionality, you can wrap the dataclass value to emulate some of the SQLAlchemy methods.

class FakeAttribute(object):
    def __init__(self, value):
        self.value = value

    def __eq__(self, value):
        return self.value == value

    def __ne__(self, value):
        return self.value != value

    def __gt__(self, value):
        return self.value > value

    def __ge__(self, value):
        return self.value >= value

    def __lt__(self, value):
        return self.value < value

    def __le__(self, value):
        return self.value <= value

    def in_(self, values):
        return self.value in values

    def not_in(self, values):
        return self.value not in values

def convert(expr, obj):
    bound_value = expr.right.value
    dataclass_value = FakeAttribute(getattr(obj, expr.left.key))
    return expr.operator(dataclass_value, bound_value)

expr = MyObjectModel.attr > 23
filtered_obj_list = [o for o in obj_list if convert(expr, o)]
print(filtered_obj_list)  # [MyObject(attr=25)]

expr = MyObjectModel.attr.in_([20, 21, 22, 23])
filtered_obj_list = [o for o in obj_list if convert(expr, o)]
print(filtered_obj_list)  # [MyObject(attr=23)]

nagataaaas · Accepted Answer · 2022-10-06 15:22:55Z

This kinda works

import dataclasses
from typing import List

from sqlalchemy import Column, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql import operators
from sqlalchemy.sql.elements import BinaryExpression, BindParameter

Base = declarative_base()


@dataclasses.dataclass
class MyObject:
    attr: int


class MyObjectModel(Base):
    __tablename__ = "object"

    attr = Column(Integer, nullable=False, primary_key=True)


def eval_binary_expression(expression: BinaryExpression, obj, eval_as=None) -> bool:
    if eval_as:
        obj = eval_as(**obj.__dict__)

    def eval_value(value):
        if isinstance(value, BinaryExpression):
            return eval_binary_expression(value, obj)
        if isinstance(value, BindParameter):
            return value.value
        if isinstance(value, Column):
            if value.table.fullname == obj.__tablename__:
                return getattr(obj, value.name)
            raise ValueError(f"Column {value.name} is not from table {eval_as.__tablename__}")
        return value

    left = eval_value(expression.left)
    right = eval_value(expression.right)
    if expression.operator == operators.in_op:
        return left in right
    return expression.operator(left, right)


def sqlalchemy_operator_to_python(operator):
    if operator == operators.in_op:
        return lambda left, right: left in right
    elif operator == operators.not_in_op:
        return lambda left, right: left not in right
    elif operator == operators.is_:
        return lambda left, right: left == right
    elif operator == operators.isnot:
        return lambda left, right: left != right
    return operator


obj_list: List[MyObject] = [MyObject(23), MyObject(42), MyObject(25)]

print([obj for obj in obj_list if eval_binary_expression(MyObjectModel.attr > 24, obj, MyObjectModel)])
# [MyObject(attr=42), MyObject(attr=25)]

print([obj for obj in obj_list if eval_binary_expression(MyObjectModel.attr.in_([23, 50]), obj, MyObjectModel)])
# [MyObject(attr=23)]

First of all, BinaryExpression is a class to make an SQL Statement so you need some workaround to apply it to Python. So, I think this is something you have to compromise.

You need to check all operator you want to use in sqlalchemy_operator_to_python, but this works.

Collectives™ on Stack Overflow

How can I apply a SqlAlchemy `BinaryExpression` as a filter on a list?

2 Answers 2

Comments

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Related