The pythonic SQLAlchemy query module provides concise, Pythonic query syntax for SQLAlchemy.

The pythonic_sqlalchemy_query module provides concise, Pythonic query syntax for SQLAlchemy. For example, these two queries produce identical results:

pythonic_query = session.User['jack'].addresses['jack@google.com']
traditional_query = (
    # Ask for the Address...
    session.query(Address).
    # by querying a User named 'jack'...
    select_from(User).filter(User.name == 'jack').
    # then joining this to the Address 'jack@google.com`.
    join(Address).filter(Address.email_address == 'jack@google.com')

Installation

pip install pythonic_sqlalchemy_query

Use with SQLAlchemy

For most cases:

from pythonic_sqlalchemy_query import QueryMakerSession

# Construct an engine as usual.
engine = create_engine(...)
# Create a session aware of this module.
Session = sessionmaker(bind=engine, class_=QueryMakerSession)
session = Session()

# After defining some declarative classes, query away:
for result in session.User['jack'].addresses:
    # Do some processing on result...

The examples provide full, working code.

Use with Flask-SQLAlchemy

For most cases:

from pythonic_sqlalchemy_query.flask import SQLAlchemyPythonicQuery

app = Flask(__name__)
db = SQLAlchemyPythonicQuery(app)

# After defining some declarative classes, query away:
for result in User['jack'].addresses:
    # Do some processing on result...

The examples provide full, working code.

Documentation

See the pythonic_sqlalchemy_query module.

License

This software is distributed under the terms of the MIT license.

pythonic_sqlalchemy_query/__init__.py - Provide concise, Pythonic query syntax for SQLAlchemy

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

flask.py - Provide extensions for Flask-SQLAlchemy

See test_flask.py - Unit tests for ../pythonic_sqlalchemy_query/flask.py for usage and examples.

Imports

These are listed in the order prescribed by PEP 8.

Standard library

None.

Third-party imports

from sqlalchemy.ext.declarative import declarative_base
from flask import _app_ctx_stack
from flask_sqlalchemy import SQLAlchemy
from flask_sqlalchemy.model import Model, DefaultMeta
 

Local imports

from . import QueryMakerScopedSession, QueryMaker
 
 

Flask-SQLAlchemy customization

Create a SQLAlchemy class that includes the pythonic_sqlalchemy_query extension. In particular, it supports the following improvements over Flask-SQLAlchemy’s queries:

• User['peter'].q.first() as a shortcut for User.query.filter_by(username='peter').first().
• db.session(User)['peter'].q.first() as a shortcut to db.session.query(User).filter_by(username='peter').first().

To use: db = SQLAlchemyPythonicQuery(app) instead of db = SQLAlchemy(app), as shown in test_flask.py - Unit tests for ../pythonic_sqlalchemy_query/flask.py.

Enable syntax such as Model[id] for queries.

class QueryMakerFlaskDeclarativeMeta(DefaultMeta):
    def __getitem__(cls, key):

Extract the session from the Flask-SQLAlchemy query attribute.

        session = cls.query.session

Build a new query from here, since cls.query has already invoked add_entity on cls.

        return QueryMaker(cls, session.query())[key]
 
 

Then, use this in the Flask-SQLAlchemy session.

class SQLAlchemyPythonicQuery(SQLAlchemy):

Unlike the standard SQLAlchemy parameters, this does not allow the model_class keyword argument.

    def __init__(self, *args, **kwargs):

Provide a declarative_base model for Flask-SQLAlchemy. This is almost identical to code in flask_sqlalchemy.SQLAlchemy.make_declarartive_base, but using our custom metaclass.

        assert 'model_class' not in kwargs
        kwargs['model_class'] = declarative_base(
            cls=Model,
            name='Model',
            metadata=kwargs.get('metadata', None),

Use our custom metaclass.

            metaclass=QueryMakerFlaskDeclarativeMeta
        )

        super(SQLAlchemyPythonicQuery, self).__init__(*args, **kwargs)
 

Enable syntax such as db.session(User)['peter']. The only change from the Flask-SQLAlchemy v2.3.2 source: QueryMakerScopedSession instead of orm.scoped_session.

    def create_scoped_session(self, options=None):
        if options is None:
            options = {}

        scopefunc = options.pop('scopefunc', _app_ctx_stack.__ident_func__)
        options.setdefault('query_cls', self.Query)
        return QueryMakerScopedSession(
            self.create_session(options), scopefunc=scopefunc
        )

Import everything from core.py - Core routines into this.

from .core import *
 

Define the version of this module.

__version__ = '1.2.0'

tests – Unit tests and example of use

test_flask.py - Unit tests for flask.py - Provide extensions for Flask-SQLAlchemy

This file provide working examples and tests for flask.py - Provide extensions for Flask-SQLAlchemy.

Imports

These are listed in the order prescribed by PEP 8.

Standard library

None

Third-party imports

from flask import Flask
from sqlalchemy.sql.expression import func
 

Local imports

from pythonic_sqlalchemy_query.flask import SQLAlchemyPythonicQuery
from util import print_query
 

Setup

Create a basic Flask-SQLAlchemy application.

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemyPythonicQuery(app)
 

Model

class User(db.Model):
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String)
    fullname = db.Column(db.String)
    password = db.Column(db.String)
 

Define a default query which assumes the key is a User’s name if given a string.

    @classmethod
    def default_query(cls, key):
        if isinstance(key, str):
            return cls.name == key

    def __repr__(self):
       return "<User(name='%s', fullname='%s', password='%s')>" % (
                            self.name, self.fullname, self.password)

class Address(db.Model):
    __tablename__ = 'addresses'
    id = db.Column(db.Integer, primary_key=True)
    email_address = db.Column(db.String, nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'))

    user = db.relationship("User", back_populates="addresses")
 

Define a default query which assumes the key is an Address’s e-mail address.

    @classmethod
    def default_query(cls, key):
        return cls.email_address == key

    def __repr__(self):
        return "<Address(email_address='%s')>" % self.email_address

User.addresses = db.relationship(
    "Address", order_by=Address.id, back_populates="user")
 

Create all tables.

db.create_all()

Test data

jack = User(name='jack', fullname='Jack Bean', password='gjffdd')
jack.addresses = [
                  Address(email_address='jack@google.com'),
                  Address(email_address='j25@yahoo.com')]
db.session.add(jack)
db.session.commit()
 

Demonstration and unit tests

Traditional versus Pythonic

def test_traditional_versus_pythonic():

Create a query to select the Address for ‘jack@google.com’ from User ‘jack’.

The Pythonic version of a query:

    pythonic_query = "User['jack'].addresses['jack@google.com']"
    print_query(pythonic_query, [jack.addresses[0]], globals())
 

The traditional approach:

    traditional_query = (

Ask for the Address…

        "Address.query."

by querying a User named ‘jack’…

        "select_from(User).filter(User.name == 'jack')."

then joining this to the Address ‘jack@google.com’.

        "join(Address).filter(Address.email_address == 'jack@google.com')"
    )
    print_query(traditional_query, [jack.addresses[0]], globals())
 

Examples

def test_more_examples():

Ask for the full User object for jack.

    print_query("User['jack']", [jack], globals())

Ask only for Jack’s full name.

    print_query("User['jack'].fullname", [(jack.fullname, )], globals())

Get all of Jack’s addresses.

    print_query("User['jack'].addresses", jack.addresses, globals())

Get just the email-address of all of Jack’s addresses.

    print_query("User['jack'].addresses.email_address", [(x.email_address, ) for x in jack.addresses], globals())

Get just the email-address j25@yahoo.com of Jack’s addresses.

    print_query("User['jack'].addresses['j25@yahoo.com']", [jack.addresses[1]], globals())

Ask for the full Address object for j25@yahoo.com.

    print_query("Address['j25@yahoo.com']", [jack.addresses[1]], globals())

Ask for the User associated with this address.

    print_query("Address['j25@yahoo.com'].user", [jack], globals())

Use a filter criterion to select a User with a full name of Jack Bean.

    print_query("User[User.fullname == 'Jack Bean']", [jack], globals())

Use two filter criteria to find the user named jack with a full name of Jack Bean.

    print_query("User['jack'][User.fullname == 'Jack Bean']", [jack], globals())

Look for the user with id 1.

    print_query("User[1]", [jack], globals())

Use an SQL expression in the query.

    print_query("User[func.lower(User.fullname) == 'jack bean']", [jack], globals())

Ask for everything in User. (This is built in to Flask-SQLAlchemy and isn’t a part of this package. However, this seems like the cleanest syntax to me.)

    print_query("User.query", [jack], globals())

Another syntax for the everything in User. This does use functionality from this package.

    print_query("db.session(User)", [jack], globals())

Ask for the name of all Users. Note that User.name can’t be used in this case – this refers to the name attribute of the User class.

    print_query("db.session(User).name", [(jack.name, )], globals())
 

Query using the session. A bit longer, but it produces the same results. For comparison:

    print_query(           " User['jack'].addresses['jack@google.com']", [jack.addresses[0]], globals())
    print_query("db.session(User)['jack'].addresses['jack@google.com']", [jack.addresses[0]], globals())
 

main

Run the example code. This can also be tested using pytest.

if __name__ == '__main__':
    test_traditional_versus_pythonic()
    test_more_examples()

test_pythonic_sqlalchemy_query.py - Unit tests and demonstrations for pythonic_sqlalchemy_query/__init__.py - Provide concise, Pythonic query syntax for SQLAlchemy

To run: execute pytest tests from the project’s root directory.

Imports

These are listed in the order prescribed by PEP 8.

Standard library

from pprint import pprint
 

Third-party imports

from sqlalchemy import create_engine, ForeignKey, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.orm import aliased
from sqlalchemy.ext import baked
from sqlalchemy.sql.expression import bindparam
from sqlalchemy.sql.expression import func
 

Local imports

from pythonic_sqlalchemy_query import (
    QueryMaker, QueryMakerDeclarativeMeta, QueryMakerQuery, QueryMakerSession
)
from util import print_query, _print_query
 

Setup

Database setup

engine = create_engine('sqlite:///:memory:')#, echo=True)

The QueryMakerSession allows syntax such as session(User).... For typical use, you may omit the query_cls=QueryMakerQuery. See sessionmaker, query_cls, and class_.

Session = sessionmaker(bind=engine, query_cls=QueryMakerQuery, class_=QueryMakerSession)
session = Session()
 

Model

Use the QueryMakerDeclarativeMeta in our declarative class definitions.

Base = declarative_base(metaclass=QueryMakerDeclarativeMeta)

Create a simple User and Adddress based on the tutorial.

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    fullname = Column(String)
    password = Column(String)
 

Define a default query which assumes the key is a User’s name if given a string.

    @classmethod
    def default_query(cls, key):
        if isinstance(key, str):
            return cls.name == key

    def __repr__(self):
       return "<User(name='%s', fullname='%s', password='%s')>" % (
                            self.name, self.fullname, self.password)

class Address(Base):
    __tablename__ = 'addresses'
    id = Column(Integer, primary_key=True)
    email_address = Column(String, nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'))

    user = relationship("User", back_populates="addresses")
 

Define a default query which assumes the key is an Address’s e-mail address.

    @classmethod
    def default_query(cls, key):
        return cls.email_address == key

    def __repr__(self):
        return "<Address(email_address='%s')>" % self.email_address

User.addresses = relationship(
    "Address", order_by=Address.id, back_populates="user")
 

Create all tables.

Base.metadata.create_all(engine)
 

Test data

jack = User(name='jack', fullname='Jack Bean', password='gjffdd')
jack.addresses = [
                  Address(email_address='jack@google.com'),
                  Address(email_address='j25@yahoo.com')]
session.add(jack)
session.commit()
 

Demonstration and unit tests

Traditional versus Pythonic

def test_traditional_versus_pythonic():

Create a query to select the Address for ‘jack@google.com’ from User ‘jack’.

The Pythonic version of a query:

    pythonic_query = "session(User)['jack'].addresses['jack@google.com']"
    print_query(pythonic_query, [jack.addresses[0]], globals())
 

The traditional approach:

    traditional_query = (

Ask for the Address…

        "session.query(Address)."

by querying a User named ‘jack’…

        "select_from(User).filter(User.name == 'jack')."

then joining this to the Address ‘jack@google.com`.

        "join(Address).filter(Address.email_address == 'jack@google.com')"
    )
    print_query(traditional_query, [jack.addresses[0]], globals())
 

More examples

def test_more_examples():

Ask for the full User object for jack.

    print_query("session(User)['jack']", [jack], globals())

Ask only for Jack’s full name.

    print_query("session(User)['jack'].fullname", [(jack.fullname, )], globals())

Get all of Jack’s addresses.

    print_query("session(User)['jack'].addresses", jack.addresses, globals())

Get just the email-address of all of Jack’s addresses.

    print_query("session(User)['jack'].addresses.email_address", [(x.email_address, ) for x in jack.addresses], globals())

Get just the email-address j25@yahoo.com of Jack’s addresses.

    print_query("session(User)['jack'].addresses['j25@yahoo.com']", [jack.addresses[1]], globals())

Ask for the full Address object for j25@yahoo.com.

    print_query("session(Address)['j25@yahoo.com']", [jack.addresses[1]], globals())

Ask for the User associated with this address.

    print_query("session(Address)['j25@yahoo.com'].user", [jack], globals())

Use a filter criterion to select a User with a full name of Jack Bean.

    print_query("session(User)[User.fullname == 'Jack Bean']", [jack], globals())

Use two filter criteria to find the user named jack with a full name of Jack Bean.

    print_query("session(User)['jack'][User.fullname == 'Jack Bean']", [jack], globals())

Look for the user with id 1.

    print_query("session(User)[1]", [jack], globals())

Use an SQL expression in the query.

    print_query("session(User)[func.lower(User.fullname) == 'jack bean']", [jack], globals())

Ask for all Users.

    print_query("session(User)", [jack], globals())

Ask for the name of all Users.

    print_query("session(User).name", [(jack.name, )], globals())
 

Transform to a query for indexing.

    assert _print_query("session(Address).q[1]", globals()) == jack.addresses[1]

Call the count method on the underlying Query object.

    assert _print_query("session(Address).q.count()", globals()) == 2

Call the order_by method on the underlying Query object.

    print_query("session(Address).q.order_by(Address.email_address)", list(reversed([jack.addresses][0])), globals())

Use the underlying query object for complex joins.

    adalias1 = aliased(Address)
    print_query("session(User).q.join(adalias1, User.addresses)['j25@yahoo.com']", [jack.addresses[1]], globals(), locals())
 

Queries are generative: qm can be re-used.

    qm = session(User)['jack']
    print_query("qm.addresses", jack.addresses, globals(), locals())
    print_query("qm", [jack], globals(), locals())
 

Properties and variables can be accessed as usual.

    cds_str = "session(User)['jack'].fullname.q.column_descriptions"
    print('-'*78)
    print('Code: {}\nResult:'.format(cds_str))
    cds = eval(cds_str)
    assert cds[0]['name'] == 'fullname'
    pprint(cds)
    print('')
 

Advanced examples

def test_advanced_examples():

Specify exactly what to return by accessing the underlying query.

    print_query("session(User)['jack'].addresses._query.add_columns(User.id, Address.id)", [(1, 1), (1, 2)], globals() )
 

If QueryMakerSession isn’t used, the session can be provided at the end of the query. However, this means the .q property won’t be useful (since it has no assigned session).

    print_query("User['jack'].to_query(session)", [jack], globals())
 

If the QueryMakerDeclarativeMeta metaclass wasn’t used, this performs the equivalent of User['jack'] manually.

    print_query("QueryMaker(User)['jack'].to_query(session)", [jack], globals())
 

Add to an existing query: first, find the User named jack.

    q = session.query().select_from(User).filter(User.name == 'jack')

Then ask for the Address for jack@google.com.

    print_query("q.query_maker().addresses['jack@google.com']", [jack.addresses[0]], globals(), locals())

Do the same manually (without relying on the QueryMakerQuery query_maker method).

    print_query("QueryMaker(query=q).addresses['jack@google.com']", [jack.addresses[0]], globals(), locals())
 

Baked queries are supported.

    bakery = baked.bakery()
    baked_query = bakery(lambda session: session(User))
    baked_query += lambda query: query[User.name == bindparam('username')]

The last item in the query must end with a .q. Note that this doesn’t print nicely. Using .to_query() instead fixes this.

    baked_query += lambda query: query.q.order_by(User.id).q
    print_query("baked_query(session).params(username='jack', email='jack@google.com')", [jack], globals(), locals())
 

main

Run the example code. This can also be tested using pytest.

if __name__ == '__main__':
    test_traditional_versus_pythonic()
    test_more_examples()
    test_advanced_examples()

util.py - Utilities for testing

Imports

These are listed in the order prescribed by PEP 8.

Standard library

None.

Third-party imports

None.

Local imports

from pythonic_sqlalchemy_query import QueryMaker
 

Code

Print a query its underlying SQL.

def _print_query(str_query, globals_, locals_=None):
    print('-'*78)
    print('Query: ' + str_query)
    query = eval(str_query, globals_, locals_)
    if isinstance(query, QueryMaker):
        query = query.q
    print('Resulting SQL emitted:\n{}\nResults:'.format(str(query)))
    return query
 

Print the results of a query and optionally compare the results with the expected value.

def print_query(str_query, expected_result=None, globals_=None, locals_=None):
    query = _print_query(str_query, globals_, locals_)
    for _ in query:
        print(_)
    print('')
    if expected_result:
        assert query.all() == expected_result
    return query

setup.py - Package and install pythonic_sqlalchemy_query

To package

Create a source distribution, a built distribution, then upload both to pythonic_sqlalchemy_query at PyPI:

python -m pip install -U pip setuptools wheel twine
python setup.py sdist bdist_wheel
python -m twine upload dist/*

For development:

pip install -e .

Packaging script

Otherwise known as the evils of setup.py.

PyPA copied code

From PyPA’s sample setup.py, read long_description from a file. This code was last updated on 26-May-2015 based on this commit.

Always prefer setuptools over distutils

from setuptools import setup

To use a consistent encoding

from os import path

Imports for version parse code.

import os
import re

here = path.abspath(path.dirname(__file__))
 

Get the long description from the relevant file.

with open(path.join(here, 'README.rst'), encoding='utf-8') as f:
    long_description = f.read()

The inclusion of a raw tag causes PyPI to not render the reST. Ouch. Remove it before uploading.

    long_description = re.sub('\.\. raw.*<\/iframe>', '', long_description, flags=re.DOTALL)
 

This code was copied from version parse code. See version in the call to setup below.

def read(*names, **kwargs):
    with open(
        os.path.join(os.path.dirname(__file__), *names),
        encoding=kwargs.get("encoding", "utf8")
    ) as fp:
        return fp.read()

def find_version(*file_paths):
    version_file = read(*file_paths)
    version_match = re.search(r"^__version__ = ['\"]([^'\"]*)['\"]",
                              version_file, re.M)
    if version_match:
        return version_match.group(1)
    raise RuntimeError("Unable to find version string.")
 

My code

setup(

This must comply with PEP 0426’s name requirements.

    name='pythonic_sqlalchemy_query',
 

Projects should comply with the version scheme specified in PEP440. I use this so that my Sphinx docs will have the same version number. There are a lot of alternatives in Single-sourcing the Project Version. While I like something simple, such as import pythonic_sqlalchemy_query then version=pythonic_sqlalchemy_query.__version__ here, this means any dependeninces of pythonic_sqlalchemy_query/__init__.py - Provide concise, Pythonic query syntax for SQLAlchemy will be requred to run setup, a bad thing. So, instead I read the file in setup.py and parse the version with a regex (see version parse code).

    version=find_version("pythonic_sqlalchemy_query/__init__.py"),

    description="Provide concise, Pythonic query syntax for SQLAlchemy",
    long_description=long_description,
 

The project’s main homepage.

    url='http://pythonic_sqlalchemy_query.readthedocs.io/en/latest/',

    author="Bryan A. Jones",
    author_email="bjones@ece.msstate.edu",

    license='GPLv3+',
 

These are taken from the full list.

    classifiers=[
        'Development Status :: 5 - Production/Stable',
        'Intended Audience :: Developers',
        'License :: OSI Approved :: GNU General Public License v3 or later (GPLv3+)',
        'Operating System :: OS Independent',
        'Programming Language :: Python :: 2',
        'Programming Language :: Python :: 2.7',
        'Programming Language :: Python :: 3',
        'Programming Language :: Python :: 3.3',
        'Programming Language :: Python :: 3.4',
        'Programming Language :: Python :: 3.5',
        'Programming Language :: Python :: 3.6',
        'Topic :: Database',
    ],

    keywords='SQLAlchemy, query helper',

    packages=['pythonic_sqlalchemy_query'],
 

List run-time dependencies here. These will be installed by pip when your project is installed. For an analysis of “install_requires” vs pip’s requirements files see: https://packaging.python.org/en/latest/requirements.html

    install_requires=([
        'SQLAlchemy',
    ]),
 

List additional groups of dependencies here (e.g. development dependencies). You can install these using the following syntax, for example:

$ pip install -e .[test]

    extras_require={
        'test': ['pytest', 'Flask-SQLAlchemy'],
    },
)

setup.cfg - Configuration for setup.py - Package and install pythonic_sqlalchemy_query

This package supports Python 2 and Python 3. Mark it as a univeral wheel.

[bdist_wheel]
universal=1

.. error:: SyntaxError: invalid syntax (line 14). Docstrings cannot be processed.

.gitignore - Files for Git to ignore

syntax: glob
 

Python compiled files

*.pyc
*.pyo
.cache/
 

CodeChat-generated files

_build
 

Files created when packaging

build
dist
 

Others

*.egg-info/
sphinx-enki-info.txt

History

• Development version:
  • No changes yet.
• 1.2.0: 11-Jan-2018
  • Changed syntax: session(User) instead of session.User to avoid “magic” method use.
  • Added additional tests.
• 1.1.1: 15-Nov-2017
  • Refactor test code.
  • Refactor Flask’s scoped session into the core.
• 1.1.0: 13-Nov-2017
  • Restructured to place code in a package.
  • Added Flask-SQLAlchemy extensions.
• 1.0.4: 6-Nov-2017
  • Changed license to MIT to match SQLAlchemy’s license.
• 1.0.3: 3-Nov-2017
  • Correct broken hyperlinks.
• 1.0.2: 3-Nov-2017
  • Provide a better __iter__ for QueryMaker.
  • Correct the __setattr__ in _QueryWrapper.
• 1.0.1: 3-Nov-2017
  • Allow access to Query variables and special methods from a _QueryWrapper.
• 1.0.0: 3-Nov-2017
  • Inital release.

Before PyPI release

This program was originally posted on the SQLAlchemy mailing list. Thanks to Mike Bayer for helpful feedback.

requirements.txt - A pip requirements file needed to build the documentation

This is needed by the ReadTheDocs build process.

CodeChat
 

CodeChat.css - Style sheet for CodeChat docs

Stylesheet for use with CodeChat’s extensions to Docutils.

The overall goal is to allow comments (everything but <pre> tags) to appear with normal (double) spacing, while comments and code are single spaced.

Author:Bryan A. Jones Contact:bjones AT ece DOT msstate DOT edu Copyright:This stylesheet has been placed in the public domain.

Implementation

This will cause an error if the html4css1 style sheet is embedded instead of linked, though the page still displays without problems. However, omitting it will cause the linked stylesheet case to fail, so leave this in.

@import url(html4css1.css);
 

Remove all top and bottom space around <pre>, making it single-spaced. Also, remove the border so that code and comments mix more naturally.

.fenced-code pre, div.fenced-code div.highlight, div.fenced-code div.highlight pre, pre.fenced-code {
    padding-top: 0px;
    padding-bottom: 0px;
    padding-left: 0px;
    margin-top: 0px;
    margin-bottom: 0px;
    margin-left: 0px;
    border: none;
    -webkit-box-shadow: none;
}
 

Tweak the size of the indent so that code and comments line up.

div.CodeChat-indent {

This is only needed for the ReadTheDocs theme.

    font-size: 10pt;
}

conf.py - Configuration file for a Sphinx/CodeChat project

This file configures Sphinx, which transforms restructured text (reST) into html. See Sphinx build configuration file docs for more information on the settings below.

This file was originally created by sphinx-quickstart, then modified by hand. Notes on its operation:

• This file is execfile()d by Sphinx with the current directory set to its containing dir.
• Not all possible configuration values are present in this autogenerated file.
• All configuration values have a default; values that are commented out serve to show the default.

import os
import sphinx_rtd_theme
import pythonic_sqlalchemy_query
 

If extensions (or modules to document with autodoc) are in another directory, add these directories to sys.path here. If the directory is relative to the documentation root, use os.path.abspath to make it absolute, as shown here.

##sys.path.insert(0, os.path.abspath('.'))

Project information

project and copyright: General information about the project. Change this for your project.

project = 'pythonic_sqlalchemy_query'
copyright = '2017, Bryan A. Jones'
 

The version info for the project you’re documenting, acts as replacement for |version| and |release|, also used in various other places throughout the built documents. Change these for your project.

version: The short X.Y version.

version = pythonic_sqlalchemy_query.__version__

release: The full version, including alpha/beta/rc tags.

release = 'version ' + version
 

There are two options for replacing |today|:

1. If you set today to some non-false value, then it is used:

##today = ''

2. Otherwise, today_fmt is used as the format for a strftime call.

##today_fmt = '%B %d, %Y'
 

highlight_language: The default language to highlight source code in.

highlight_language = 'python3'
 

pygments_style: The style name to use for Pygments highlighting of source code.

pygments_style = 'sphinx'
 

add_function_parentheses: If true, ‘()’ will be appended to :func: etc. cross-reference text.

##add_function_parentheses = True
 

add_module_names: If true, the current module name will be prepended to all description unit titles (such as .. function::).

##add_module_names = True
 

show_authors: If true, sectionauthor and moduleauthor directives will be shown in the output. They are ignored by default.

##show_authors = False
 

modindex_common_prefix: A list of ignored prefixes for module index sorting.

##modindex_common_prefix = []
 

General configuration

extensions: If your documentation needs a minimal Sphinx version, state it here.

##needs_sphinx = '1.5'
 

Add any Sphinx extension module names here, as strings. They can be extensions coming with Sphinx (named ‘sphinx.ext.*’) or your custom ones. CodeChat note: The CodeChat.CodeToRestSphinx extension is mandatory; without it, CodeChat will not translate source code to reST and then (via Sphinx) to html.

extensions = ['CodeChat.CodeToRestSphinx']
 

templates_path: Add any paths that contain templates here, relative to this directory.

templates_path = ['_templates']
 

rst_epilog: A string of reStructuredText that will be included at the end of every source file that is read.

rst_epilog = (

Provide a convenient way to refer to a source file’s name.

"""

.. |docname| replace:: :docname:`name`
"""
)
 

source_suffix: The suffix of source filenames.

source_suffix = '.rst'
 

CodeChat note: A dict of {glob, lexer_alias}, which uses lexer_alias (e.g. a lexer’s short name) to analyze any file wihch matches the given glob-style pattern (e.g. glob.

CodeChat_lexer_for_glob = {

CSS files are auto-detected as a CSS + Lasso file by Pygments, causing it to display incorrectly. Define them as CSS only.

    '*.css': 'CSS',

These files use # for comments, so treat them like Python.

    '*.txt': 'Python',
    '*.gitignore': 'Python',
}
 
 

CodeChat note: CodeChat_excludes is a list of exclude_patterns which applies only to source documents; exclude_patterns will exclude the given files from all of Sphinx (for example, files here won’t be included even if they’re mentioned in html_static_path.

CodeChat_excludes = []
 

source_encoding: The encoding of source files.

##source_encoding = 'utf-8-sig'
 

master_doc: The master toctree document.

master_doc = 'index'
 

language: The language for content autogenerated by Sphinx. Refer to documentation for a list of supported languages.

##language = None
 

exclude_patterns: List of patterns, relative to source directory, that match files and directories to ignore when looking for source files.

Important: Do NOT add CodeChat.css to this list; this will instruct Sphinx not to copy it to the _static directory, where it is needed to properly lay out CodeChat output. Instead, to exclude it from the documents produced by Sphinx, add it to CodeChat_excludes.

exclude_patterns = [

CodeChat notes:

By default, Enki will instruct Sphinx to place all Sphinx output in _build; this directory should therefore be excluded from the list of source files.

    '_build',

The CodeToRestSphinx extension creates a file named sphinx-enki-info.txt, which should be ignored by Sphinx.

    'sphinx-enki-info.txt',

Python files.

    '*.egg-info',
    'build',
    'dist',
    '__pycache__',
    '.cache',
]
 

default_role: The reST default role (used for this markup: `text`) to use for all documents.

default_role = 'any'
 

keep_warnings: If true, keep warnings as “system message” paragraphs in the built documents. Regardless of this setting, warnings are always written to the standard error stream when sphinx-build is run. CodeChat note: This should always be True; doing so places warnings next to the offending text in the web view, making them easy to find and fix.

keep_warnings = True

Options for HTML output

html_theme: The theme to use for HTML and HTML Help pages.

html_theme = 'sphinx_rtd_theme'
 

html_theme_options: Theme options are theme-specific and customize the look and feel of a theme further.

##html_theme_options = {}
 

html_style: The style sheet to use for HTML pages.

##html_style = None
 

html_theme_path: Add any paths that contain custom themes here, relative to this directory.

html_theme_path = [sphinx_rtd_theme.get_html_theme_path()]
 

html_title: The name for this set of Sphinx documents. If None, it defaults to <project> v<release> documentation.

##html_title = None
 

html_short_title: A shorter title for the navigation bar. Default is the same as html_title.

##html_short_title = None
 

html_logo: The name of an image file (relative to this directory) to place at the top of the sidebar.

##html_logo = None
 

html_favicon: The name of an image file (within the static path) to use as favicon of the docs. This file should be a Windows icon file (.ico) being 16x16 or 32x32 pixels large.

##html_favicon = None
 

html_static_path: Add any paths that contain custom static files (such as style sheets) here, relative to this directory. They are copied after the builtin static files, so a file named default.css will overwrite the builtin default.css. CodeChat note: This must always include CodeChat.css.

html_static_path = ['CodeChat.css']
 

html_last_updated_fmt: If not ‘’, a ‘Last updated on:’ timestamp is inserted at every page bottom, using the given strftime format.

html_last_updated_fmt = '%b, %d, %Y'
 

html_sidebars: Custom sidebar templates, maps document names to template names.

##html_sidebars = {}
 

html_additional_pages: Additional templates that should be rendered to pages, maps page names to template names.

##html_additional_pages = {}
 

html_domain_indices: If false, no module index is generated.

##html_domain_indices = True
 

html_use_index: If false, no index is generated.

##html_use_index = True
 

html_split_index: If true, the index is split into individual pages for each letter.

##html_split_index = False
 

html_copy_source: If true, the reST sources are included in the HTML build as _sources/name.

html_copy_source = True
 

html_show_sourcelink: If true, links to the reST sources are added to the pages.

html_show_sourcelink = True
 

html_sourcelink_suffix: Suffix to be appended to source links (see html_show_sourcelink), unless they have this suffix already.

html_sourcelink_suffix = ''
 

html_show_sphinx: If true, “Created using Sphinx” is shown in the HTML footer. Default is True.

##html_show_sphinx = True
 

html_show_copyright: If true, “(C) Copyright …” is shown in the HTML footer. Default is True.

##html_show_copyright = True
 

html_use_opensearch: If true, an OpenSearch description file will be output, and all pages will contain a <link> tag referring to it. The value of this option must be the base URL from which the finished HTML is served.

##html_use_opensearch = ''
 

html_file_suffix: This is the file name suffix for HTML files (e.g. “.xhtml”).

##html_file_suffix = None





###########################################################################

auto-created readthedocs.org specific configuration #

###########################################################################
 
 

The following code was added during an automated build on readthedocs.org It is auto created and injected for every build. The result is based on the conf.py.tmpl file found in the readthedocs.org codebase: https://github.com/rtfd/readthedocs.org/blob/master/readthedocs/doc_builder/templates/doc_builder/conf.py.tmpl

 
 
import sys
import os.path
from six import string_types

from sphinx import version_info
 

Get suffix for proper linking to GitHub This is deprecated in Sphinx 1.3+, as each page can have its own suffix

if globals().get('source_suffix', False):
    if isinstance(source_suffix, string_types):
        SUFFIX = source_suffix
    else:
        SUFFIX = source_suffix[0]
else:
    SUFFIX = '.rst'
 

Add RTD Static Path. Add to the end because it overwrites previous files.

if not 'html_static_path' in globals():
    html_static_path = []
if os.path.exists('_static'):
    html_static_path.append('_static')
html_static_path.append('/home/docs/checkouts/readthedocs.org/readthedocs/templates/sphinx/_static')
 

Add RTD Theme only if they aren’t overriding it already

using_rtd_theme = False
if 'html_theme' in globals():
    if html_theme in ['default']:

Allow people to bail with a hack of having an html_style

        if not 'html_style' in globals():
            import sphinx_rtd_theme
            html_theme = 'sphinx_rtd_theme'
            html_style = None
            html_theme_options = {}
            if 'html_theme_path' in globals():
                html_theme_path.append(sphinx_rtd_theme.get_html_theme_path())
            else:
                html_theme_path = [sphinx_rtd_theme.get_html_theme_path()]

            using_rtd_theme = True
else:
    import sphinx_rtd_theme
    html_theme = 'sphinx_rtd_theme'
    html_style = None
    html_theme_options = {}
    if 'html_theme_path' in globals():
        html_theme_path.append(sphinx_rtd_theme.get_html_theme_path())
    else:
        html_theme_path = [sphinx_rtd_theme.get_html_theme_path()]
    using_rtd_theme = True

if globals().get('websupport2_base_url', False):
    websupport2_base_url = 'https://readthedocs.org/websupport'
    if 'http' not in settings.MEDIA_URL:
        websupport2_static_url = 'https://media.readthedocs.org/static/'
    else:
        websupport2_static_url = 'https://media.readthedocs.org//static'


#Add project information to the template context.
context = {
    'using_theme': using_rtd_theme,
    'html_theme': html_theme,
    'current_version': "latest",
    'version_slug': "latest",
    'MEDIA_URL': "https://media.readthedocs.org/",
    'PRODUCTION_DOMAIN': "readthedocs.org",
    'versions': [
    ("latest", "/en/latest/"),
    ("stable", "/en/stable/"),
    ],
    'downloads': [
    ("htmlzip", "//readthedocs.org/projects/pythonic-sqlalchemy-query/downloads/htmlzip/latest/"),
    ],
    'subprojects': [
    ],
    'slug': 'pythonic-sqlalchemy-query',
    'name': u'Pythonic SQLAlchemy query',
    'rtd_language': u'en',
    'canonical_url': 'http://pythonic-sqlalchemy-query.readthedocs.io/en/latest/',
    'analytics_code': '',
    'single_version': False,
    'conf_py_path': '/./',
    'api_host': 'https://readthedocs.org',
    'github_user': 'bjones1',
    'github_repo': 'pythonic_sqlalchemy_query',
    'github_version': 'master',
    'display_github': True,
    'bitbucket_user': 'None',
    'bitbucket_repo': 'None',
    'bitbucket_version': 'master',
    'display_bitbucket': False,
    'gitlab_user': 'None',
    'gitlab_repo': 'None',
    'gitlab_version': 'master',
    'display_gitlab': False,
    'READTHEDOCS': True,
    'using_theme': (html_theme == "default"),
    'new_theme': (html_theme == "sphinx_rtd_theme"),
    'source_suffix': SUFFIX,
    'user_analytics_code': '',
    'global_analytics_code': 'UA-17997319-1',

    'commit': '01a71b02',

}




if 'html_context' in globals():
    html_context.update(context)
else:
    html_context = context
 

Add custom RTD extension

if 'extensions' in globals():
    extensions.append("readthedocs_ext.readthedocs")
else:
    extensions = ["readthedocs_ext.readthedocs"]

License

Copyright 2017 Bryan A. Jones

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

License

This software is distributed under the terms of the MIT license.