Тип хинтинг sqlalchemy результат запроса - PullRequest
Новая
       30

Тип хинтинг sqlalchemy результат запроса

2 голосов
/ 26 марта 2019

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

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

Тип каждого объекта в записях выглядит как sqlalchemy.util._collections.result, но быстрый 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.

1 Ответ

1 голос
/ 27 марта 2019

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

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

...