输入提示sqlalchemy查询结果

Posted 2021-04-25

我无法弄清楚sqlalchemy查询返回什么类型的对象。

entries = session.query(Foo.id, Foo.date).all()

条目中每个对象的类型似乎是sqlalchemy.util._collections.result，但python解释器中的快速from sqlalchemy.util._collections import result引发了ImportError。

我最终要做的是输入提示这个函数：

def my_super_function(session: Session) -> ???:
    entries = session.query(Foo.id, Foo.date).all()
    return entries

我应该用什么代替???？ mypy（在这种情况下）似乎与List[Tuple[int, str]]很好，因为是的确我可以访问我的条目，如果它们是元组，但我也可以使用entry.date访问它们，例如。

答案

我也很好奇这个课程无法导入。答案很长，因为我已经告诉你我是如何解决这个问题的，请耐心等待。

Query.all()在list()对象上调用Query：

def all(self):
    """Return the results represented by this ``Query`` as a list.
    This results in an execution of the underlying query.
    """
    return list(self)

...列表将迭代对象，所以Query.__iter__()：

def __iter__(self):
    context = self._compile_context()
    context.statement.use_labels = True
    if self._autoflush and not self._populate_existing:
        self.session._autoflush()
    return self._execute_and_instances(context)

...返回Query._execute_and_instances()方法的结果：

def _execute_and_instances(self, querycontext):
    conn = self._get_bind_args(
        querycontext, self._connection_from_session, close_with_result=True
    )

    result = conn.execute(querycontext.statement, self._params)
    return loading.instances(querycontext.query, result, querycontext)

执行查询并返回sqlalchemy.loading.instances()函数的结果。在该函数中，有this line适用于非单实体查询：

keyed_tuple = util.lightweight_named_tuple("result", labels)

...如果我在那条线之后贴上print(keyed_tuple)，它会打印<class 'sqlalchemy.util._collections.result'>，这就是你在上面提到的类型。所以无论那个对象是什么，它来自sqlalchemy.util._collections.lightweight_named_tuple()函数：

def lightweight_named_tuple(name, fields):
    hash_ = (name,) + tuple(fields)
    tp_cls = _lw_tuples.get(hash_)
    if tp_cls:
        return tp_cls

    tp_cls = type(
        name,
        (_LW,),
        dict(
            [
                (field, _property_getters[idx])
                for idx, field in enumerate(fields)
                if field is not None
            ]
            + [("__slots__", ())]
        ),
    )

    tp_cls._real_fields = fields
    tp_cls._fields = tuple([f for f in fields if f is not None])

    _lw_tuples[hash_] = tp_cls
    return tp_cls

所以关键部分是this statement：

tp_cls = type(
    name,
    (_LW,),
    dict(
        [
            (field, _property_getters[idx])
            for idx, field in enumerate(fields)
            if field is not None
        ]
        + [("__slots__", ())]
    ),
)

...根据文档调用内置的type()类：

使用三个参数，返回一个新类型对象。这实际上是类语句的动态形式。

这就是为什么你不能导入类sqlalchemy.util._collections.result - 因为该类只在查询时构造。我要说的原因是，在执行查询之前，列名称（即命名的元组属性）是未知的。

来自python docs的type签名是：type(name, bases, dict)其中：

名称字符串是类名，并成为__name__属性;基元元组列出基类并成为__bases__属性;并且dict字典是包含类主体定义的命名空间，并被复制到标准字典以成为__dict__属性。

正如你所看到的，在bases传递给type()的lightweight_named_tuple()论证是(_LW,)。因此，任何动态创建的命名元组类型都继承自sqlalchemy.util._collections._LW，这是一个可以导入的类：

from sqlalchemy.util._collections import _LW

entries = session.query(Foo.id, Foo.date).all()
for entry in entries:
    assert isinstance(entry, _LW)  # True

...所以我不确定将函数键入带有前导下划线的内部类是否是好的形式，但_LW继承自sqlalchemy.util._collections.AbstractKeyedTuple，后者本身继承自tuple。这就是你当前输入List[Tuple[int, str]]的原因，因为它是一个元组列表。所以，你的选择，_LW，AbstractKeyedTuple，tuple都将是你的函数返回的正确表示。