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()