python - sqlalchemy 按表名和过滤条件过滤列表

Question

我正在使用 sqlalchemy ORM 层与 RDS 通信。 这是所有表过滤行的通用功能。 我们传递表名、要选择的列、过滤器和日期范围。

filter = { "company_guid": "xxxx", "status": "有效"}

filter 是一个字典，其中键作为列名，值是条件。 哪个工作正常

但我想过滤 status 列，其中值可以是 Active 或 TempInActive

所以现在过滤器变成了filter = { "company_guid": "xxxx", "status": ["Active", "TempActive"} 它不起作用，因为值是列表，而不是字符串。 我知道可以使用 result = session.query(Customers).filter(Customers.id.in_([1,3])) 但在我的场景中表名和列名是函数参数。

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]

谁能帮我解决这个问题？

一种方法是构造查询字符串并进行评估，但这不是好方法。

Answer 1

当您使用 ORM 时，我假设您在函数中调用的 table 实际上是一个映射的 ORM 类。

如果我理解正确，您希望能够处理 filters 的值可能是标量值的两种情况，在这种情况下您希望按相等性进行过滤，或者值列表，在这种情况下，您希望使用 in_() 测试是否存在于列表中。然而，一个复杂的因素是您不能在 filter() 中直接使用 filters 的 str 键。希望我已经理解了。

我认为你可以巧妙地解决这个问题，方法是使用 getattr 从表对象中获取列属性，条件列表理解，然后将列表解包到 .filter()，例如:

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()
        ]
    )
)

如果 val 不是 list，这将产生等同于 orm_table_object.column_attrib == val 和 orm_table_object.column_attrib .in_(val) 如果 val 是一个 list。