键入提示sqlalchemy查询结果

Question

我不知道sqlalchemy查询返回哪种对象。

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

条目中每个对象的类型似乎是sqlalchemy.util._collections.result，但是from sqlalchemy.util._collections import resultpython解释器中的快速提示会引发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()电话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)

... 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()函数的结果。在该函数中有这一行适用于非单一实体查询：

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

所以关键部分是这个声明：

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 文档中，签名type是：type(name, bases, dict)其中：

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

如您所见，bases传递给type()in的参数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都是您的函数返回内容的正确表示。