python - SQLAlchemy cascading polymorphic column updates -

i have parent employee table , child engineer table. client perspective want interact employee model. implemented read , delete, issues arise when trying update or insert.

the sqlalchemy docs state:

warning

currently, 1 discriminator column may set, typically on base-most class in hierarchy. “cascading” polymorphic columns not yet supported.

so seem default not going work. i'm looking ideas on how make work.

here's complete test setup using postgres psycopg2. sql might work other sql databases, have test others.

sql script create test database (testdb) , tables (employee, engineer):

create database testdb; \c testdb;  create table employee(   id    int   primary key   not null,   name  text,   type  text   );  create table engineer(   id            int   primary key   not null,   engineer_name text,   employee_id int references employee(id)     on update cascade     on delete cascade   ); 

python test script:

as-is insert test fail, delete pass. if change code (comment/uncomment) use child engineer model pass both cases.

import sqlalchemy sa import sqlalchemy.orm orm sqlalchemy.ext.declarative import declarative_base sqlalchemy import (     column,     foreignkey,     integer,     text,     )  base = declarative_base()  class employee(base):     __tablename__ = 'employee'     id = column(integer, primary_key=true)     name = column(text(), default='john')     type = column(text, default='engineer')      __mapper_args__ = {         'polymorphic_identity':'employee',         'polymorphic_on':type,         'with_polymorphic': '*',     }  class engineer(employee):     __tablename__ = 'engineer'     id = column(integer, foreignkey('employee.id',                 ondelete='cascade', onupdate='cascade'), primary_key=true)     engineer_name = column(text(), default='eugine')      __mapper_args__ = {         'polymorphic_identity':'engineer',     }   def count(session, model):   query = session.query(model)   count = len(query.all())   return count  url = 'postgresql+psycopg2://postgres@localhost/testdb' engine = sa.create_engine(url) base.metadata.bind = engine base.metadata.create_all() session = orm.sessionmaker(engine) session = session()  if __name__ == '__main__':   id=0   print '#'*30, 'insert', '#'*30   id += id   # want interact employee table   e = employee(id=id)   # use child model see insert test pass   # e = engineer(id=id)   session.add(e)   session.commit()   print 'pass' if count(session, employee) == count(session, engineer) else 'fail'    print '#'*30, 'delete', '#'*30   # e = session.query(employee).first()   session.delete(e);   session.commit();   print 'pass' if count(session, employee) == count(session, engineer) else 'fail'    session.flush() 

any ideas on how accomplish through sqlalchemy model definitions without having use explicit controller code?

thanks!

edit i'm not getting love one. have ideas on how accomplish with controller code?

using controller logic can accomplished getting polymorphic subclass using polymorphic identity.

i'm adding 2 functions encapsulate basic logic.

def get_polymorphic_class(klass, data):   column = klass.__mapper__.polymorphic_on   if column none:     # column not polymorphic class can returned as-is     return klass   identity = data.get(column.name)   if not identity:     raise valueerror('missing value "' + column.name + '"', data)   mapper = klass.__mapper__.polymorphic_map.get(identity)   if mapper:     return mapper.class_   else:     raise valueerror('missing polymorphic_identity definition "' + identity + '"')   return klass  def insert(klass, data):   klass = get_polymorphic_class(klass, data)   e = klass(**data)   session.add(e)   session.commit()   return e 

now update main use insert function , works expected:

if __name__ == '__main__':   id=0   print '#'*30, 'insert', '#'*30   id += id   e = insert(employee, {'id': id, 'type': 'engineer'})   print 'pass' if count(session, employee) == count(session, engineer) else 'fail'    print '#'*30, 'delete', '#'*30   session.delete(e);   session.commit();   print 'pass' if count(session, employee) == count(session, engineer) else 'fail'   session.flush() 

there's code in encapsulation reusability, important part doing employee.__mapper__.polymorphic_map['engineer'].class_ returns engineer class can proper cascading insert.