类型提示 sqlalchemy 查询结果

Question

我不知道 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 访问它们，例如。

Answer 1

我也对无法导入该类感到好奇。答案很长，我已经向您介绍了我是如何解决的，请耐心等待。

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

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)

... list 将迭代对象，所以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字符串是类名，成为__name__属性； 基元组逐项列出基类并成为__bases__ 属性;并且 dict 字典是包含的命名空间 类主体的定义并被复制到标准字典中 成为__dict__ 属性。

如您所见，在lightweight_named_tuple() 中传递给type() 的bases 参数是(_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，而sqlalchemy.util._collections.AbstractKeyedTuple 本身继承自tuple。这就是您当前键入List[Tuple[int, str]] 有效的原因，因为它是 元组列表。因此，请自行选择，_LW、AbstractKeyedTuple、tuple 都将正确表示您的函数返回的内容。