时间戳

首页 > 解决方案 > sqlalchemy 按表名过滤列表并过滤条件

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

问题描述

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

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

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]

谁能帮我解决这个问题？

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

标签： pythonsqldatabasesqlalchemy

解决方案

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

如果我理解正确，您希望能够处理值filters可能是标量值（在这种情况下您希望过滤相等性）或值列表（在这种情况下您希望）的两种情况使用 测试列表中的存在in_()。但是，一个复杂的因素是您不能直接使用 in的str键。希望我已经明白了。filtersfilter()

我认为您可以通过使用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()
        ]
    )
)

这将产生orm_table_object.column_attrib == valif valis not alist和orm_table_object.column_attrib.in_(val)if valis a的等价物list。

推荐阅读