Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

Type hinting sqlalchemy query result

Tags:

I can't figure out what kind of object a sqlalchemy query returns.

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

The type of each object in entries seems to be sqlalchemy.util._collections.result, but a quick from sqlalchemy.util._collections import result in a python interpreter raises an ImportError.

What I'm ultimately trying to do is to type hint this function:

def my_super_function(session: Session) -> ???:     entries = session.query(Foo.id, Foo.date).all()     return entries 

What should I put in place of ???? mypy (in this case) seems to be fine with List[Tuple[int, str]] because yes indeed I can access my entries like if they were tuples, but i can also access them with entry.date, for example.

like image 865
JPFrancoia Avatar asked Mar 26 '19 15:03

JPFrancoia


People also ask

What does SQLAlchemy query return?

It returns exactly one result or raise an exception. It applies one or more ORDER BY criterion to the query and returns the newly resulting Query. It performs a bulk update query and updates rows matched by this query in the database.

Does SQLAlchemy cache results?

SQLAlchemy supports two types of caches: Caching the result set so that repeatedly running the same query hits the cache instead of the database. It uses dogpile which supports many different backends, including memcached , redis , and basic flat files.

What does query all () return?

all() will return all records which match our query as a list of objects.


1 Answers

I too found it curious that the class couldn't be imported. The answer is pretty long as I've walked you through how I've worked it out, bear with me.

Query.all() calls list() on the Query object itself:

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) 

... where list will be iterating over the object, so 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) 

... returns the result of Query._execute_and_instances() method:

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) 

Which executes the query and returns the result of sqlalchemy.loading.instances() function. In that function there is this line which applies to non-single-entity queries:

keyed_tuple = util.lightweight_named_tuple("result", labels) 

... and if I stick a print(keyed_tuple) in after that line it prints <class 'sqlalchemy.util._collections.result'>, which is the type that you mention above. So whatever that object is, it's coming from the sqlalchemy.util._collections.lightweight_named_tuple() function:

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 

So the key part is this statement:

tp_cls = type(     name,     (_LW,),     dict(         [             (field, _property_getters[idx])             for idx, field in enumerate(fields)             if field is not None         ]         + [("__slots__", ())]     ), ) 

... which calls the built in type() class which according to the docs:

With three arguments, return a new type object. This is essentially a dynamic form of the class statement.

And this is why you cannot import the class sqlalchemy.util._collections.result - because the class is only constructed at query time. I'd say that the reason for this is that the column names (i.e. the named tuple attributes) aren't known until the query is executed).

From python docs the signature for type is: type(name, bases, dict) where:

The name string is the class name and becomes the __name__ attribute; the bases tuple itemizes the base classes and becomes the __bases__ attribute; and the dict dictionary is the namespace containing definitions for class body and is copied to a standard dictionary to become the __dict__ attribute.

As you can see, the bases argument passed to type() in lightweight_named_tuple() is (_LW,). So any of the dynamically created named tuple types inherit from sqlalchemy.util._collections._LW, which is a class that you can import:

from sqlalchemy.util._collections import _LW  entries = session.query(Foo.id, Foo.date).all() for entry in entries:     assert isinstance(entry, _LW)  # True 

... so I'm not sure whether it's good form to type your function to an internal class with the leading underscore, but _LW inherits from sqlalchemy.util._collections.AbstractKeyedTuple, which itself inherits from tuple. That's why your current typing of List[Tuple[int, str]] works, because it is a list of tuples. So take your pick, _LW, AbstractKeyedTuple, tuple would all be correct representations of what your function is returning.

like image 93
SuperShoot Avatar answered Sep 30 '22 16:09

SuperShoot