I tried doing this, it seems to kind of work:
class TestTask(Task): """Test task - only for use in tests.""" __tablename__ = None # suuper needed without it polymorphic filter will not work task_type_name = TaskTypeEnum.test_worker_task.value @task def process(self): """Process test task, just mark it as completed.""" self.completed = True db.session.commit() return True def init_type_map(engine): """Update polymorphic map to use task type IDs instead of names. Call before doing anything. """ # skip if we don't have the task type table yet if not engine.has_table(TaskType.__tablename__): return mapper = Task.__mapper__ poly_map = mapper.self_and_descendants # load a list of all task names from "task_type_name" class properties task_type_names = [] task_id_mapper_map = dict() for task_mapper in mapper.self_and_descendants: if task_mapper is mapper: continue # this is _task, not a descendant task_class = task_mapper.class_ if not hasattr(task_class, 'task_type_name'): log.warning(f"Not loading task {task_class}; missing task_type_name") continue task_type_names.append(task_class.task_type_name) task_id_mapper_map[task_class.task_type_name] = task_mapper if not task_type_names: # no tasks with type name set return # upsert task names to make sure they all have IDs rows = [{'task_type': name} for name in task_type_names] insert_query = pg_insert(TaskType).on_conflict_do_nothing(index_elements =['task_type']).values(rows) conn = engine.connect() Session = sessionmaker(bind=engine) session = Session(bind=conn) session.execute(insert_query) session.commit() task_id_type_map = get_type_map(session) conn.close() # reverse task_type_id_map = {v:k for k, v in task_id_type_map.items()} for type_name in task_type_names: if type_name in task_id_type_map: # already mapped this ID continue task_mapper = task_id_mapper_map[type_name] task_type_id = task_type_id_map[type_name] # this should exist now # change mapping to be task_id=>mapper instead of task_type=>mapper mapper.polymorphic_map.update({task_type_id: task_mapper}) task_mapper.polymorphic_identity = task_type_id def get_type_map(session): """Get id/name mapping of task types.""" task_type_id_map_res = session.execute(f"SELECT id,task_type FROM task_type") task_id_type_map = dict() for r in task_type_id_map_res: task_id_type_map[r[0]] = r[1] return task_id_type_map Is this on this right path? On Monday, November 27, 2017 at 4:50:04 AM UTC+13, Mike Bayer wrote: > > On Sun, Nov 26, 2017 at 6:19 AM, Mischa S <mis...@jetbridge.com > <javascript:>> wrote: > > I want to have a polymorphic table that gets its identity from a column > in a > > "type" table. I have a "task" table and a "task_type" table. I want to > be > > able to create task subclasses that say something like: > > > > __mapper_args__ = { > > 'polymorphic_identity': 'do_stuff_task', > > } > > > > (Assuming that a row with do_stuff_task is created in the task_type > table) > > > > > > Currently in my base Task class I do: > > > > task_type_name = > > column_property(select([TaskType.task_type]).where(TaskType.id == > > task_type_id).as_scalar()) > > > > __mapper_args__ = { > > 'polymorphic_on': task_type_name, > > } > > > this looks like you got this from > > https://bitbucket.org/zzzeek/sqlalchemy/wiki/UsageRecipes/PolymorphicOnAnotherTable, > > > yup > > > > > > > > If I try to get say, a user.do_stuff_tasks relationship it emits the > > following SQL: > > > > SELECT > > (SELECT task_type.task_type > > FROM task_type > > WHERE task_type.id = task.task_type_id) > > FROM task > > WHERE 123 = task.user_id > > AND > > (SELECT task_type.task_type > > FROM task_type > > WHERE task_type.id = task.task_type_id) IN ('do_stuff_task'); > > > > > > This is technically correct, however this query is highly problematic > for > > us. It does a scan on task in the subquery, which is extremely > unpleasant > > and basically takes forever. > > What I want is something more like: > > > > SELECT * > > FROM task st > > JOIN task_type stt ON st.task_type_id=stt.id > > WHERE stt.task_type='do_stuff_task' > > AND st.user_id=123; > > > > Is there some way to rewrite the column_property or polymorphic identity > > condition to combine the WHERE conditions in the JOIN instead of doing > two > > subselects? > > this would be very difficult to generalize. the > polymorphic_discriminator is only specified as a column expression and > in SQLAlchemy a column expression never implies a JOIN in the query, a > JOIN is always an explicit add on Query. You could potentially > implement the before_compile() event for Query and add this JOIN > manually, but this would be very difficult to generalize to all cases. > Here's how to do it for the demo on the wiki mentioned above: > > from sqlalchemy import event > from sqlalchemy.orm.query import Query > > @event.listens_for(Query, "before_compile", retval=True) > def setup_query_for_poly(query): > entities = set(d['entity'] for d in query.column_descriptions) > if A in entities: > query = query.join(AType) > return query > > > with the mapping like: > > class A(Base): > __tablename__ = 'a' > > id = Column(Integer, primary_key=True) > data = Column(String) > type_id = Column(ForeignKey('atype.id')) > type_name = column_property(AType.name) > type = relationship(AType) > > __mapper_args__ = { > "polymorphic_on": type_name, > "polymorphic_identity": "a" > } > > > then you get a query like this: > > SELECT atype.name AS atype_name, a.id AS a_id, a.data AS a_data, > a.type_id AS a_type_id > FROM a JOIN atype ON atype.id = a.type_id > > > but....if your query refers to "AType" elsewhere, or multiple times, > or you refer to "A" in different contexts, adding a JOIN like this > could get in the way. Or if "A" is represented using aliasing, you'd > need to dig more carefully into query.column_descriptions and apply > the join() more carefully. > > The other way to do this is just to put the polymorphic identities > into the mapper._polymorphic_map: > > A.__mapper__.polymorphic_map.update( > (key, A.__mapper__.polymorphic_map[value]) > for (key, value) in sess.query(AType.id, AType.name) > ) > > with mapping like: > > class A(Base): > __tablename__ = 'a' > > id = Column(Integer, primary_key=True) > data = Column(String) > type_id = Column(ForeignKey('atype.id')) > type = relationship(AType) > > __mapper_args__ = { > "polymorphic_on": type_id, > "polymorphic_identity": "a" > } > > > > > > > > > > > > Thanks! > > > > -- > > 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:>. > > To post to this group, send email to 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.