core.py - Core routines

Imports

These are listed in the order prescribed by PEP 8.

Standard library

None.

Third-party imports

from sqlalchemy.orm import Query, scoped_session
from sqlalchemy.orm.attributes import InstrumentedAttribute
from sqlalchemy.orm.properties import ColumnProperty, RelationshipProperty
from sqlalchemy.ext.declarative import DeclarativeMeta
from sqlalchemy.orm.base import _generative
from sqlalchemy.sql.elements import ClauseElement
from sqlalchemy.orm.session import Session
from sqlalchemy.orm.util import class_mapper
from sqlalchemy.orm.mapper import Mapper
from sqlalchemy.inspection import inspect
 
 

QueryMaker

This class provides a concise, Pythonic syntax for simple queries; as shown in the Demonstration and unit tests, session(User)['jack'].addresses produces a Query for the Address of a User named jack.

This class provides the following methods:

• Constructor: session(User) (with help from QueryMakerSession) creates a query on a User table.
• Indexing: session(User)['jack'] performs filtering.
• Attributes: session(User)['jack'].addresses joins to the Addresses table.
• Iteration: for x in session(User)['jack'].addresses iterates over the results of the query.
• Query access: User['jack'].addresses.q returns a Query-like object. Any Query method can be invoked on it.

See the Demonstration and unit tests for examples and some less-used methods.

This works by translating class instances in a query/select, indexes into filters, and columns/relationships into joins. The following code shows the Pythonic syntax on the first line, followed by the resulting translation into SQLAlchemy performed by this class on the next line.

session(User)                     ['jack']                   .addresses
session.query().select_from(User).filter(User.name == 'jack').join(Address).add_entity(Address)

Limitations

Note that the delete and update methods cannot be invoked on the query produced by this class. Safer (but lower-performance) is:

for _ in session(User)['jack']:
    session.delete(_)

Rationale:

• Per the docs on delete and update, these come with a long list of caveats. Making dangerous functions easy to invoke is poor design.
• For implementation, QueryMaker cannot invoke select_from. Doing so raises sqlalchemy.exc.InvalidRequestError: Can't call Query.update() or Query.delete() when join(), outerjoin(), select_from(), or from_self() has been called. So, select_from must be deferred – but to when? User['jack'].addresses requires a select_from, while User['jack'] needs just add_entity. We can’t know which to invoke until the entire expression is complete.

class QueryMaker(object):
    def __init__(self,

An optional Declarative class to query.

        declarative_class=None,

Optionally, begin with an existing query.

        query=None):

        if declarative_class:
            assert _is_mapped_class(declarative_class)
 

If a query is provided, try to infer the declarative_class.

        if query is not None:
            assert isinstance(query, Query)
            self._query = query
            try:
                self._select = self._get_joinpoint_zero_class()
            except:

We can’t infer it. Use what’s provided instead, and add this to the query.

                assert declarative_class
                self._select = declarative_class
                self._query = self._query.select_from(declarative_class)
            else:

If a declarative_class was provided, make sure it’s consistent with the inferred class.

                if declarative_class:
                    assert declarative_class is self._select
        else:

The declarative class must be provided if the query wasn’t.

            assert declarative_class

Since a query was not provied, create an empty query; to_query will fill in the missing information.

            self._query = Query([]).select_from(declarative_class)

Keep track of the last selectable construct, to generate the select in to_query.

            self._select = declarative_class
 

Copied verbatim from sqlalchemy.orm.query.Query._clone. This adds the support needed for the generative interface. (Mostly) quoting from query, “QueryMaker features a generative interface whereby successive calls return a new QueryMaker object, a copy of the former with additional criteria and options associated with it.”

    def _clone(self):
        cls = self.__class__
        q = cls.__new__(cls)
        q.__dict__ = self.__dict__.copy()
        return q
 

Looking up a class’s Column or relationship generates the matching query.

    @_generative()
    def __getattr__(self, name):

Find the Column or relationship in the join point class we’re querying.

        attr = getattr(self._get_joinpoint_zero_class(), name)

If the attribute refers to a column, save this as a possible select statement. Note that a Column gets replaced with an InstrumentedAttribute; see QueryableAttribute.

        if isinstance(attr.property, ColumnProperty):
            self._select = attr
        elif isinstance(attr.property, RelationshipProperty):

Figure out what class this relationship refers to. See mapper.params.class_.

            declarative_class = attr.property.mapper.class_

Update the query by performing the implied join.

            self._query = self._query.join(declarative_class)

Save this relationship as a possible select statement.

            self._select = declarative_class
        else:

This isn’t a Column or a relationship.

            assert False
 

Indexing the object performs the implied filter. For example, session(User)['jack'] implies session.query(User).filter(User.name == 'jack').

    @_generative()
    def __getitem__(self,

Most often, this is a key which will be filtered by the default_query method of the currently-active Declarative class. In the example above, the User class must define a default_query to operate on strings. However, it may also be a filter criterion, such as session(User)[User.name == 'jack'].

        key):
 

See if this is a filter criterion; if not, rely in the default_query defined by the Declarative class or fall back to the first primary key.

        criteria = None
        jp0_class = self._get_joinpoint_zero_class()
        if isinstance(key, ClauseElement):
            criteria = key
        elif hasattr(jp0_class, 'default_query'):
            criteria = jp0_class.default_query(key)
        if criteria is None:
            pks = inspect(jp0_class).primary_key
            criteria = pks[0] == key
        self._query = self._query.filter(criteria)
 

Support common syntax: for x in query_maker: converts this to a query and returns results. The session must already have been set.

    def __iter__(self):
        return self.to_query().__iter__()
 

This property returns a _QueryWrapper, a query-like object which transforms returned Query values back into this class while leaving other return values unchanged.

    @property
    def q(self):
        return _QueryWrapper(self)
 

Transform this object into a Query.

    def to_query(self,

Optionally, the Session to run this query in.

        session=None):
 

If a session was specified, use it to produce the query; otherwise, use the existing query.

        query = self._query.with_session(session) if session else self._query

Choose the correct method to select either a column or a class (e.g. an entity). As noted earlier, a Column becomes and InstrumentedAttribute.

        if isinstance(self._select, InstrumentedAttribute):
            return query.add_columns(self._select)
        else:
            return query.add_entity(self._select)
 

Get the right-most join point in the current query.

    def _get_joinpoint_zero_class(self):
        jp0 = self._query._joinpoint_zero()

If the join point was returned as a Mapper, get the underlying class.

        if isinstance(jp0, Mapper):
            jp0 = jp0.class_
        return jp0
 
 

_QueryWrapper

This class behaves mostly like a Query. However, if the return value of a method is a Query, it returns a QueryMaker object instead. It’s intended for internal use by QueryMaker.q.

class _QueryWrapper(object):
    def __init__(self, query_maker):
        self._query_maker = query_maker
 

Delegate directly to the wrapped Query. Per special method lookup, the special method names bypass __getattr__ (and even __getattribute__) lookup. Only override what Query overrides.

The _tq (to_query) property shortens the following functions.

    @property
    def _tq(self):
        return self._query_maker.to_query()
    def __getitem__(self, key):
        return self._tq.__getitem__(key)
    def __str__(self):
        return self._tq.__str__()
    def __repr__(self):
        return self._tq.__repr__()
    def __iter__(self):
        return self._tq.__iter__()
 

Allow __init__ to create the _query_maker variable. Everything else goes to the wrapped Query. Allow direct assignments, as this mimics what an actual Query instance would do.

    def __setattr__(self, name, value):
        if name != '_query_maker':
            return self._query_maker.__setattr__(name, value)
        else:
            self.__dict__[name] = value
 

Run the method on the underlying Query. If a Query is returned, wrap it in a QueryMaker.

    def __getattr__(self, name):
        attr = getattr(self._tq, name)
        if not callable(attr):

If this isn’t a function, then don’t do any wrapping.

            return attr
        else:
            def _wrap_query(*args, **kwargs):

Invoke the requested Query method on the “completed” query returned by to_query.

                ret = attr(*args, **kwargs)
                if isinstance(ret, Query):

If the return value was a Query, make it generative by returning a new QueryMaker instance wrapping the query.

                    query_maker = self._query_maker._clone()

Re-run getattr on the raw query, since we don’t want to add columns or entities to the query yet. Otherwise, they’d be added twice (here and again when to_query is called).

                    query_maker._query = getattr(query_maker._query, name)(*args, **kwargs)

If the query involved a join, then the join point has changed. Update what to select.

                    query_maker._select = query_maker._get_joinpoint_zero_class()
                    return query_maker
                else:

Otherwise, just return the result.

                    return ret

            return _wrap_query
 
 

QueryMakerDeclarativeMeta

Turn indexing of a Declarative class into a query. For example, User['jack'] is a query. See the Advanced examples for an example of its use.

class QueryMakerDeclarativeMeta(DeclarativeMeta):
    def __getitem__(cls, key):
        return QueryMaker(cls)[key]
 
 

QueryMakerQuery

Provide support for changing a Query instance into a QueryMaker instance. See the Advanced examples for an example of its use.

TODO: This doesn’t allow a user-specified Query class. Perhaps provide a factory instead?

class QueryMakerQuery(Query):
    def query_maker(self, declarative_class=None):
        return QueryMaker(declarative_class, self)
 
 

QueryMakerSession

Create a Session which returns a QueryMaker when called as a function. This enables session(User)['jack']. See the Database setup for an example of its use.

class QueryMakerSession(Session):
    def __call__(self, declarative_class):
        return QueryMaker(declarative_class, self.query())
 
 

QueryMakerScopedSession

Provide QueryMakerSession extensions for a scoped session.

class QueryMakerScopedSession(scoped_session):

Note that the superclass’ __call__ method only accepts keyword arguments. So, only return a QueryMaker if only arguments, not keyword arguments, are given.

    def __call__(self, *args, **kwargs):
        if args and not kwargs:
            return QueryMaker(*args, query=self.registry().query())
        else:
            return super().__call__(*args, **kwargs)
 
 

Support routines

Copied from https://stackoverflow.com/a/7662943.

def _is_mapped_class(cls):
    try:
        class_mapper(cls)
        return True
    except:
        return False