from sqlalchemy import create_engine, ForeignKey, Column, Integer, String from sqlalchemy.ext.declarative import declarative_base, DeclarativeMeta from sqlalchemy.orm import sessionmaker, Query, relationship from sqlalchemy.orm.attributes import InstrumentedAttribute from sqlalchemy.orm.properties import ColumnProperty, RelationshipProperty from sqlalchemy.orm.base import _generative from sqlalchemy.sql.elements import ClauseElement
This class provides a concise, Pythonic syntax for simple queries; for example, User['jack'].addresses produces a Query for the Address of a User named jack. Comparing this to a traditional query:
User ['jack'] .addresses Query([]).select_from(User).filter(User.name == 'jack').join(Address).add_entity(Address)
See the demonstration for more examples.
class QueryMaker:
def __init__(self,
A Declarative class to query.
declarative_class,
Optionally, begin with an existing query.
query=None):
Keep track of which Declarative class we're querying.
self.declarative_class = declarative_class
Keep track of the last selectable construct, to generate the select in go.
self.select = declarative_class
If it's not provied, create an empty query; go will fill in the missing information. TODO: If a query was provided, could I infer the declarative_class based on what the left side of a join would be? There's a reset_joinpoint, but I want to find the current joinpoint. There's a _joinpoint _joinpoint_zero(), but I don't understand these very well.
self.query = query or Query([]).select_from(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 Declarative class we're querying.
attr = getattr(self.declarative_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
else:
assert isinstance(attr.property, RelationshipProperty)
Figure out what class this relationship refers to. See mapper.params.class_.
self.declarative_class = attr.property.mapper.class_
Update the query by performing the implied join.
self.query = self.query.join(self.declarative_class)
Save this relationship as a possible select statement.
self.select = self.declarative_class
Indexing the object performs the implied filter. For example, User['jack'] implies query.filter(User.name == 'jack').
@_generative()
def __getitem__(self,
Mostly 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 User[User.name == 'jack'].
key):
See if this is a filter criterion; if not, rely in the default_query defined by the Declarative class.
criteria = key if isinstance(key, ClauseElement) else self.declarative_class.default_query(key)
self.query = self.query.filter(criteria)
Transform this object into a Query.
def to_query(self,
Optionally, the Session to run this query in.
session=None):
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)
Turn indexing of a Declarative class into a query. For example, User['jack'] is a query. See the model for an example of its use.
class QueryMakerDeclarativeMeta(DeclarativeMeta):
def __getitem__(cls, key):
return QueryMaker(cls)[key]
Provide support for changing a Query instance into a QueryMaker instance.See the database setup for an example of its use.
class QueryMakerQuery(Query):
def query_maker(self, declarative_class):
return QueryMaker(declarative_class, self)
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 unless the key is an int; in that case, assume the key is an id.
@classmethod
def default_query(cls, key):
return cls.id == key if isinstance(key, int) else 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")
engine = create_engine('sqlite:///:memory:')#, echo=True)
See sessionmaker, query_cls.
Session = sessionmaker(bind=engine, query_cls=QueryMakerQuery) session = Session() Base.metadata.create_all(engine)
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()
Print the results of a query.
def print_query(str_query):
print('-'*78)
print('Query: ' + str_query)
query = eval(str_query)
print('Resulting SQL emitted:\n{}\nResults:'.format(str(query)))
for _ in query:
print(_)
print()
return query
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'].to_query(session)
print_query('pythonic_query')
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')
Ask for the full User object for jack.
print_query("User['jack'].to_query(session)")
Ask only for Jack's full name.
print_query("User['jack'].fullname.to_query(session)")
Get all of Jack's addresses.
print_query("User['jack'].addresses.to_query(session)")
Get just the email-address of all of Jack's addresses.
print_query("User['jack'].addresses.email_address.to_query(session)")
Get just the email-address j25@yahoo.com of Jack's addresses.
print_query("User['jack'].addresses['j25@yahoo.com'].to_query(session)")
Ask for the full Address object for j25@yahoo.com.
print_query("Address['j25@yahoo.com'].to_query(session)")
Ask for the User associated with this address.
print_query("Address['j25@yahoo.com'].user.to_query(session)")
Use a filter criterion to select a User with a full name of Jack Bean.
print_query("User[User.fullname == 'Jack Bean'].to_query(session)")
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'].to_query(session)")
Look for the user with id 1.
print_query("User[1].to_query(session)")
Queries are generative: q can be re-used.
q = User['jack']
print_query("q.addresses.to_query(session)")
print_query("q.to_query(session)")
Add additional filters to the query.
print_query("User['jack'].addresses.to_query(session).filter(Address.email_address == 'jack@google.com')")
If the QueryMakerDeclarativeMeta metaclass wasn't used, this performs the equivalent of User['jack'] manually.
print_query("QueryMaker(User)['jack'].to_query(session)")
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. Note that no session needs to be provided to go
print_query("q.query_maker(User).addresses['jack@google.com'].to_query()")
Do the same manually (without relying on the QueryMakerQuery query_maker method).
print_query("QueryMaker(User, q).addresses['jack@google.com'].to_query()")