CODEWITHSUNDEEP
pythonsqldatabasesqlalchemy
Vivek Sable
Vivek Sable

Reputation: 10213

sqlalchemy to filter on list by table name and filters condition

I am using sqlalchemy ORM layer to communicate with RDS. Here this is common function for all tables to filter row. We pass table name, column to select, filters and date range.

filter = { "company_guid": "xxxx", "status": "Active"}

filter is dictionary which have key as column name and value is condition. Which work fine

but know I want filter on status column where value can be Active or TempInActive

So now filter become filter = { "company_guid": "xxxx", "status": ["Active", "TempActive"} It not working because value is list, not string. I know can use result = session.query(Customers).filter(Customers.id.in_([1,3])) but in my scenario table name and column names are function arguments. 

def get_items_withvalue(self, table_name, column_name=None, attribute_value=None,
                        columns_to_select=None, date_value=False, filters=None):
    """
    @Summary: This method used to get data based on a condition.
    @param table_name (string): This is the table_name
    @param column_name (None/string): for the column_name
    @param attribute_value (None/list/string): for the column_value
    @params columns_to_select(None/list/string): columns to send in response
    @params filters(None/dict): where clause for rows to be fetched
    @return (list of dict): fetched rows or count from DB
    """
    data = []
    session = None
    try:
        # Get session which communicate with RDS
        session = self.get_session()
        table = str_to_class(table_name)
        if columns_to_select:
            data = []
        else:
            if isinstance(attribute_value, list):
                data = (session.query(table)
                        .filter(getattr(table, column_name)
                                .in_(attribute_value))
                        .all())
            elif date_value:
                data = (session.query(table)
                        .filter(cast(getattr(table, column_name), Date)
                                == attribute_value)
                        .all())
            elif filters:
                ## How to update following code to filter on list(in)
                # filters is dictionary
                data = (session.query(table).filter_by(**filters).all())
                ##
            else:
                data = (
                    session.query(table).filter(getattr(table, column_name)
                                                == attribute_value).all()
                )
    except Exception as err:
        self.logger.exception("Error fetching items ")
        raise Exception(err)
    finally:
        if session:
            session.close()
    if columns_to_select:
        return [row._asdict() for row in data]
    return [object_as_dict(row) for row in data]

Can anyone help me to solve this?

One way is to construct query string and do eval, but which is not good way.

Upvotes: 1

Views: 3033

Answers (2)

technazi
technazi

Reputation: 958

Extending the above answer just with a list of conditions

filters = [orm_table_object.field_name_1 == expected_value_1]
if expected_value_2 is not None:
   filters.append(orm_table_object.field_name_2 == expected_value_2)
if expected_value_3 is not None:
   filters.append(orm_table_object.field_name_3 == expected_value_3)


session.query(table).filter(
    *[f for f in filters]
)

Upvotes: 1

SuperShoot
SuperShoot

Reputation: 10861

As you are using the ORM I'll assume that what you call table in the function is actually a mapped ORM class.

If I understand correctly, you want to be able to handle both cases where the values of filters may be either a scalar value, in which case you'd like to filter on equality, or a list of values, in which case you'd like to test for presence in the list using in_(). However, a complicating factor is that you cannot use the str keys of filters directly in filter(). Hopefully I've understood.

I think you can solve this neatly by using getattr to get the column attribs from the table object, a conditional list comprehension and then unpacking the list into .filter(), for example:

filters = {'a': 'scalar', 'and': ['collection', 'of', 'values']}
(
    session.query(table).filter(
        *[
            getattr(table, k).in_(v)
            if isinstance(v, list)
            else getattr(table, k) == v
            for k, v in filters.items()
        ]
    )
)

This will produce the equivalent of orm_table_object.column_attrib == val if val is not a list, and orm_table_object.column_attrib.in_(val) if val is a list.

Upvotes: 3

Related Questions