Hi all,
We have a small function that helps us create a simple search query by
automatically joining on required relations if needed.
For example, consider an employee ORM that has a 1:M relationship with
addresses (for postal/physical). We can do something like:
query = Employee().search('streetname', [Employee.name,
Address.street1])
We have that working, but when we add a second search field on Address:
query = Employee.search('streetname', [Employee.name, Address.street1,
Address.street2])
our method fails with: table name "address" specified more than once
We need to be able to identify if the query already has a join on
'address'
I've tried getting details on the query object (it
has ._from, ._from_obj, ._from_alias and .from_statement) that looked
interesting, but they don't appear to give us what we need.
Here is a cut down sample implementation that will hopefully remove any
confusion... Note the TODO: in Employee.search()
-----------------------8<-----------------------8<-----------------------8<
from sqlalchemy import create_engine, Column, ForeignKey, or_
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker, relationship,
joinedload
from sqlalchemy.types import Integer, String, Text
from sqlalchemy.sql.expression import cast
engine = create_engine('sqlite:///:memory:', echo=True)
Base = declarative_base(bind=engine)
Session = scoped_session(sessionmaker(bind=engine))
class Employee(Base):
__tablename__ = 'employee'
id = Column(Integer, primary_key=True)
name = Column(String)
def search(self, value, columns):
query = Session.query(Employee)
for i, column in enumerate(columns):
model = column.parententity.class_
if Employee is not model:
#TODO: Are we already joined from Employee onto model?
query = query.outerjoin(model)
args = [cast(c, Text).ilike('%%%s%%' % value) for c in columns]
return query.filter(or_(*args))
class Address(Base):
__tablename__ = 'address'
id = Column(Integer, primary_key=True)
employee_id = Column(Integer, ForeignKey(Employee.id))
street1 = Column(String(50))
street2 = Column(String(50))
employee = relationship(Employee)
Base.metadata.create_all()
#e = Employee(name='Bob')
#a = Address(employee=e, street1='street1', street2='street2')
#Session.add(a)
#Session.commit()
q = Employee().search('stree', [Employee.name, Address.street1,
Address.street2])
print q
"""
SELECT employee.id AS employee_id, employee.name AS employee_name
FROM employee LEFT OUTER JOIN address ON employee.id =
address.employee_id LEFT OUTER JOIN address ON employee.id =
address.employee_id
WHERE lower(CAST(employee.name AS TEXT)) LIKE lower(?) OR
lower(CAST(address.street1 AS TEXT)) LIKE lower(?) OR
lower(CAST(address.street2 AS TEXT)) LIKE lower(?)
"""
-----------------------8<-----------------------8<-----------------------8<
TIA
Jim
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to sqlalch...@googlegroups.com.
To unsubscribe from this group, send email to
sqlalchemy+unsubscr...@googlegroups.com.
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en.
