输入提示sqlalchemy查询结果

问题描述 投票:1回答:1

我无法弄清楚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访问它们,例如。

python sqlalchemy mypy
1个回答
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)

...列表将迭代对象,所以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 docstype签名是: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]]的原因,因为它是一个元组列表。所以,你的选择,_LWAbstractKeyedTupletuple都将是你的函数返回的正确表示。

© www.soinside.com 2019 - 2024. All rights reserved.