Введите подсказку результата запроса sqlalchemy

Я не могу понять, какой объект возвращает запрос sqlalchemy.

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

Тип каждого объекта в записях кажется sqlalchemy.util._collections.result, но быстрый from sqlalchemy.util._collections import result в интерпретаторе Python вызывает 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
person JPFrancoia    schedule 26.03.2019    source источник

Ответы (1)


arrow_upward
22
arrow_downward

Мне тоже было любопытно, что класс нельзя импортировать. Ответ довольно длинный, поскольку я рассказал вам, как я это сделал, терпите меня.

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() функция. В этой функции есть эта строка который применяется к запросам, не относящимся к одному объекту:

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__; кортеж bases перечисляет базовые классы и становится атрибутом __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 будут правильным представлением того, что возвращает ваша функция.

person SuperShoot    schedule 27.03.2019
comment
Вот почему нам нужны протоколы. - person Ilja Everilä; 27.03.2019
comment
List[Tuple[int, str]] может вызвать проблемы в дальнейшем, если вы собираетесь использовать метод _asdict() для каждого элемента результата: Cannot access member "_asdict" for type "Tuple[str, str]" - person pmsoltani; 01.09.2020