I've attached a file which is a variant to the employees example with
two objectives.
1. Base polymorphic_identity on select criteria (no type column).
2. Use two levels of inheritance.
The first objective seems to be met, but the second is not working
properly. I put in two Managers, two Generic Engineers and two Chemical
Engineers (which inherit from Engineer). When I select from employees,
I get eight records. The Chemical Engineers are included once as
Chemical Engineers and once and Generic Engineers.
How might this be better written to meet these objectives?
Randall
Michael Bayer wrote:
> just FYI, the "type" column idea is taken from Hibernate, and that's
> all Hibernate supports as far as polymorphic loading.
>
> But for polymorphic loading in SA, you are free to make any kind of
> "polymorphic_union" you like that can add in a functionally-generated
> "type" column, and specify it into select_table. im pretty sure this
> should work completely right now, such as:
>
> import sqlalchemy.sql as sql
>
> person_join = polymorphic_union(
> {
> 'engineer':sql.select([people.join(engineers),
> sql.column("'engineer'").label('type')]),
> 'manager':sql.select([people.join(managers),
> sql.column("'manager'").label('type')]),
> }
> )
>
> etc.
>
>
> >
>
--~--~---------~--~----~------------~-------~--~----~
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to [email protected]
To unsubscribe from this group, send email to [EMAIL PROTECTED]
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en
-~----------~----~----~----~------~----~------~--~---
from sqlalchemy import *
metadata = DynamicMetaData('testdata')
employees = Table('employees', metadata,
Column('employee_id', Integer, primary_key=True),
Column('name', String(50)),
Column('manager_data', String(50)),
Column('engineer_info', String(50)),
Column('cheme_info', String(50)),
)
managers = select([employees], employees.c.manager_data !=
None).alias('managers')
engineers = select([employees], employees.c.engineer_info !=
None).alias('engineers')
chemical_engineers = select([employees],
and_(employees.c.engineer_info != None,
employees.c.cheme_info != None)).alias(
'chemical_engineers')
class Employee(object):
def __init__(self, name):
self.name = name
def __repr__(self):
return self.__class__.__name__ + " " + self.name
class Manager(Employee):
def __init__(self, name, manager_data):
self.name = name
self.manager_data = manager_data
def __repr__(self):
return self.__class__.__name__ + " " + self.name + " " + self.manager_data
class Engineer(Employee):
def __init__(self, name, engineer_info):
self.name = name
self.engineer_info = engineer_info
def __repr__(self):
return self.__class__.__name__ + " " + self.name + " " + self.engineer_info
class ChemicalEngineer(Engineer):
def __init__(self, name, engineer_info, cheme_info):
self.name = name
self.engineer_info = engineer_info
self.cheme_info = cheme_info
def __repr__(self):
return self.__class__.__name__ + " " + self.name + " " + self.engineer_info
p_union = polymorphic_union(
{
'engineer': engineers,
'manager': managers,
'chemical_engineer': chemical_engineers
},
'type'
)
employee_mapper = mapper(Employee, p_union, polymorphic_on=p_union.c.type)
manager_mapper = mapper(Manager, managers, inherits=employee_mapper,
concrete=True, polymorphic_identity='manager')
engineer_mapper = mapper(Engineer, engineers, inherits=employee_mapper,
concrete=True, polymorphic_identity='engineer')
mapper(ChemicalEngineer, chemical_engineers, inherits=engineer_mapper,
concrete=True, polymorphic_identity='chemical_engineer')
def populate(session):
m1 = Manager('manager1', 'manager1')
m2 = Manager('manager2', 'manager2')
e1 = Engineer('engineer1', 'engineer1')
e2 = Engineer('engineer2', 'engineer2')
ce1 = ChemicalEngineer('cengineer1', 'cengineer1', 'cengineer1')
ce2 = ChemicalEngineer('cengineer2', 'cengineer2', 'cengineer2')
for o in (m1, m2, e1, e2, ce1, ce2):
session.save(o)
session.flush()
if __name__ == '__main__':
engine = create_engine('sqlite:///test.db')
engine.echo = True
metadata.connect(engine)
metadata.drop_all()
metadata.create_all()
session = create_session(engine)
populate(session)
print session.query(Employee).select()
