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 <[email protected]
> <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 [email protected] <javascript:>.
> > To post to this group, send email to [email protected]
> <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 [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
