That was the first route I tried. with_polymorphic() seems to cater to or
assume joined table inheritance. When I pass a selectable, it always ends
up *joining *my base to that selectable instead of *using only my
selectable*.
My problem might be that I'm trying to take advantage of
polymorphic_union() to render my union with all the CAST(NULL AS ...), but
it does exactly what I need for building a select on the fly.
Attached script is my failing attempt.
Is there a recipe or example using concrete inheritance and
with_polymorphic?
On Thursday, April 13, 2017 at 10:43:15 AM UTC-4, Mike Bayer wrote:
>
>
>
> On 04/13/2017 10:24 AM, Kent wrote:
> > Suppose we have the documentation's example of *Concrete Table
> > Inheritance, *where
> >
> > session.query(Employee).all()
> >
> >
> > produces this:
> >
> > SELECT pjoin.type AS pjoin_type,
> > pjoin.manager_data AS pjoin_manager_data,
> > pjoin.employee_id AS pjoin_employee_id,
> > pjoin.name AS pjoin_name, pjoin.engineer_info AS pjoin_engineer_info
> > FROM (
> > SELECT employees.employee_id AS employee_id,
> > CAST(NULL AS VARCHAR(50)) AS manager_data, employees.name AS
> name,
> > CAST(NULL AS VARCHAR(50)) AS engineer_info, 'employee' AS type
> > FROM employees
> > UNION ALL
> > SELECT managers.employee_id AS employee_id,
> > managers.manager_data AS manager_data, managers.name AS name,
> > CAST(NULL AS VARCHAR(50)) AS engineer_info, 'manager' AS type
> > FROM managers
> > UNION ALL
> > SELECT engineers.employee_id AS employee_id,
> > CAST(NULL AS VARCHAR(50)) AS manager_data, engineers.name AS
> name,
> > engineers.engineer_info AS engineer_info, 'engineer' AS type
> > FROM engineers
> > ) AS pjoin
> >
> >
> > Suppose we want to*filter certain managers*, which we can do with:
> >
> > session.query(Employee)\
> >
> > .filter(or_(
> >
> > Employee.manager_data == u'whatineed',
> >
> > Employee.manager_data == None))\
> >
> > .all()
> >
> >
> > If manager_data is indexed, many databases can no longer use this index.
> > What we really want is:
> >
> > SELECT pjoin.type AS pjoin_type,
> > pjoin.manager_data AS pjoin_manager_data,
> > pjoin.employee_id AS pjoin_employee_id,
> > pjoin.name AS pjoin_name, pjoin.engineer_info AS pjoin_engineer_info
> > FROM (
> > SELECT employees.employee_id AS employee_id,
> > CAST(NULL AS VARCHAR(50)) AS manager_data, employees.name AS
> name,
> > CAST(NULL AS VARCHAR(50)) AS engineer_info, 'employee' AS type
> > FROM employees
> > UNION ALL
> > SELECT managers.employee_id AS employee_id,
> > managers.manager_data AS manager_data, managers.name AS name,
> > CAST(NULL AS VARCHAR(50)) AS engineer_info, 'manager' AS type
> > FROM managers
> >
> > *WHERE manager_data = 'whatineed'*
> > UNION ALL
> > SELECT engineers.employee_id AS employee_id,
> > CAST(NULL AS VARCHAR(50)) AS manager_data, engineers.name AS
> name,
> > engineers.engineer_info AS engineer_info, 'engineer' AS type
> > FROM engineers
> > ) AS pjoin
> >
> >
> > Is there a way to accomplish this?
>
>
> Certainly, construct the complete UNION query that you want, most likely
> using Core select() and union(), and supply it to Query using
> with_polymorphic; see the "custom selectable" example in
>
> http://docs.sqlalchemy.org/en/latest/orm/inheritance.html#basic-control-of-which-tables-are-queried.
>
>
>
> Automating this process, perhaps you could compose some enhanced version
> of the polymorhic_union() feature that accepts additional criteria.
>
> However, I would suggest that if the "manager_data is NULL" part is what
> screws up the index, you might want to query like this instead:
>
> session.query(Employee)\
> .filter(or_(Employee.manager_data == u'whatineed',
> pjoin.c.type != 'manager')).all()
>
>
> >
> > Thanks in advance,
> > Kent
> >
> > --
> > SQLAlchemy -
> > The Python SQL Toolkit and Object Relational Mapper
> >
> > http://www.sqlalchemy.org/
> >
> > To post example code, please provide an MCVE: Minimal, Complete, and
> > Verifiable Example. See http://stackoverflow.com/help/mcve for a full
> > description.
> > ---
> > You received this message because you are subscribed to the Google
> > Groups "sqlalchemy" group.
> > To unsubscribe from this group and stop receiving emails from it, send
> > an email to sqlalchemy+...@googlegroups.com <javascript:>
> > <mailto:sqlalchemy+unsubscr...@googlegroups.com <javascript:>>.
> > To post to this group, send email to sqlal...@googlegroups.com
> <javascript:>
> > <mailto:sqlal...@googlegroups.com <javascript:>>.
> > Visit this group at https://groups.google.com/group/sqlalchemy.
> > For more options, visit https://groups.google.com/d/optout.
>
--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
http://www.sqlalchemy.org/
To post example code, please provide an MCVE: Minimal, Complete, and Verifiable
Example. See http://stackoverflow.com/help/mcve for a full description.
---
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to sqlalchemy+unsubscr...@googlegroups.com.
To post to this group, send email to sqlalchemy@googlegroups.com.
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
from sqlalchemy import *
from sqlalchemy.orm import *
engine = create_engine('sqlite:///', echo=True)
metadata = MetaData(engine)
session = sessionmaker(bind=engine)()
globalpreferences_table = Table("globalpreferences", metadata,
Column("preferenceid", Unicode(255), primary_key=True),
Column("value", Unicode(255))
)
sitepreferences_table = Table("sitepreferences", metadata,
Column("preferenceid", Unicode(255), primary_key=True),
Column("siteid", Unicode(255), primary_key=True),
Column("value", Unicode(255))
)
userpreferences_table = Table("userpreferences", metadata,
Column("preferenceid", Unicode(255), primary_key=True),
Column("username", Unicode(255), primary_key=True),
Column("value", Unicode(255))
)
def repr_attrs(obj, *attrs):
"""
build a safe __repr__ string with obj.__dict__.get() so we don't
accidentally set sqla attributes by referencing them
"""
return '<%s: ' % obj.__class__.__name__ + \
' '.join('{0[%s]}=[{1[%s]}]' % (i,i) for i in range(len(attrs)))\
.format(attrs, map(obj.__dict__.get, attrs)) + ">"
class PreferenceBase(object):
@classmethod
def query(cls):
"""
return a query
"""
return session.query(cls)
def __init__(self, prefid, val, **attrs):
self.preferenceid = prefid
self.value = val
self.__dict__.update(attrs)
class GlobalPreference(PreferenceBase):
def __repr__(self):
return repr_attrs(self,'preferenceid','value')
class SitePreference(PreferenceBase):
def __repr__(self):
return repr_attrs(self,'preferenceid','siteid','value')
class UserPreference(PreferenceBase):
def __repr__(self):
return repr_attrs(self,'preferenceid','username','value')
preferences_union = polymorphic_union({
'user': userpreferences_table,
'site': sitepreferences_table,
'global': globalpreferences_table
},
'type',
'preferences_union')
preferencebase_mapper = mapper(PreferenceBase, preferences_union,
polymorphic_on=preferences_union.c.type
)
mapper(GlobalPreference, globalpreferences_table,
inherits=preferencebase_mapper,
concrete=True,
polymorphic_identity='global'
)
mapper(SitePreference, sitepreferences_table,
inherits=preferencebase_mapper,
concrete=True,
polymorphic_identity='site',
)
mapper(UserPreference, userpreferences_table,
inherits=preferencebase_mapper,
concrete=True,
polymorphic_identity='user',
)
metadata.create_all(engine)
map(session.add, [
GlobalPreference('prefA', 'gvalA'),
GlobalPreference('prefB', 'gvalB'),
GlobalPreference('prefC', 'gvalC'),
SitePreference('prefB', 'svalB', siteid='00'),
UserPreference('prefC', 'svalC', username='kb'),
UserPreference('prefA', 'uvalA', username='kb'),
UserPreference('prefB', 'uvalB', username='kb'),
UserPreference('prefC', 'uvalC', username='jb'),
])
session.flush()
session.expunge_all()
u = polymorphic_union({
'user': userpreferences_table.select().where(userpreferences_table.c.username=='kb'),
'site': sitepreferences_table.select().where(sitepreferences_table.c.siteid=='00'),
'global': globalpreferences_table
},
'type')
a = with_polymorphic(PreferenceBase, '*', selectable=u, polymorphic_on=u.c.type)
print session.query(a)
