SQLAlchemy 复杂的多态映射和深度 Table 继承
SQLAlchemy Complex Polymorphic Mapping And Deep Table Inheritance
我有以下 table 结构（我尽可能地简化了它，缩小了 child/inheriting table 的范围 [还有其他的] 并从中删除了所有不相关的列提供的 tables):
## Base is my declarative_base
class AbstractQuestion(Base):
questionTypeId: Column = Column(
Integer, ForeignKey("luQuestionTypes.id"), index=True, nullable=False
)
__mapper_args__ = {
"polymorphic_identity": 0,
"polymorphic_on": questionTypeId,
}
class MultiChoiceQuestion(AbstractQuestion):
id: Column = Column(Integer, ForeignKey(AbstractQuestion.id), primary_key=True)
__mapper_args__ = {"polymorphic_identity": 1}
class AbstractSurveyQuestion(AbstractQuestion):
id: Column = Column(Integer, ForeignKey(AbstractQuestion.id), primary_key=True)
surveyQuestionTypeId: Column = Column(
Integer, ForeignKey("luSurveyQuestionTypes.id"), index=True, nullable=False
)
__mapper_args__ = {"polymorphic_identity": 2}
class RatingQuestion(AbstractSurveyQuestion):
id: Column = Column(
Integer, ForeignKey(AbstractSurveyQuestion.id), primary_key=True
)
我面临的挑战是，我正在尝试使 AbstractSurveyQuestion 具有两种类型的多态映射 - 一种作为 AbstractQuestion 的子级，具有 polymorphic_identity匹配 questionTypeId，但我还需要它有一个单独的 polymorphic_on 映射器用于它自己的子 table，即 RatingQuestion。
我能找到的最接近的东西是 ，但它似乎并没有完全针对我正在寻找的东西。
我还查看了 official docs about inheritance，但还是找不到我想要实现的目标的准确示例。
谁能帮我解决这个问题？
谢谢！
我在 SQLAlchemy 的 GitHub 存储库上发布了同样的问题。从维护者那里得到了这个答案：
https://github.com/sqlalchemy/sqlalchemy/discussions/8089#discussioncomment-2878725
下面的内容我也贴一下:
it sounds like you are looking for mult-level polymorphic_on. We don't support that right now without workarounds, and that's #2555 which is a feature we're unlikely to implement, or if we did it would be a long time from now.
It looks like you are using joined inheritance....so...two ways. The more SQL efficient one is to have an extra "supplemetary" column on your base table that can discriminate for AbstractSurveyQuestion...because if you query for all the AbstractQuestion objects, by default it's just going to query that one table, and needs to know from each row if that row is in fact a RatingQuestion.
the more convoluted way is to use mapper-configured with_polymorphic so that all queries for AbstractQuestion include all the tables (or a subset of tables, can be configured, but at minimum you'd need to join out to AbstractSurveyQuestion) using a LEFT OUTER JOIN (or if you really wanted to go crazy it can be a UNION ALL).
the workarounds are a little ugly since it's not very easy to get a "composed" value out of two columns in SQL, but they are contained to the base classes. Below examples work on SQLite and might need tweaking for other databases.
Here's the discriminator on base table demo, a query here looks like:
SELECT aq.id AS aq_id, aq.d1 AS aq_d1, aq.d2 AS aq_d2, CAST(aq.d1 AS VARCHAR) || ? || CAST(coalesce(aq.d2, ?) AS VARCHAR) AS _sa_polymorphic_on
FROM aq
from typing import Tuple, Optional
from sqlalchemy import cast
from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import event
from sqlalchemy import ForeignKey
from sqlalchemy import inspect
from sqlalchemy import Integer, func
from sqlalchemy import String
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import Session
Base = declarative_base()
class ident_(str):
"""describe a composed identity.
Using a string for easy conversion to a string SQL composition.
"""
_tup: Tuple[int, Optional[int]]
def __new__(cls, d1, d2=None):
self = super().__new__(cls, f"{d1}, {d2 or ''}")
self._tup = d1, d2
return self
def _as_tuple(self):
return self._tup
class AbstractQuestion(Base):
__tablename__ = "aq"
id = Column(Integer, primary_key=True)
d1 = Column(
Integer, nullable=False
) # this can be your FK to the other table etc.
d2 = Column(
Integer, nullable=True
) # this is a "supplementary" discrim column
__mapper_args__ = {
"polymorphic_identity": ident_(0),
"polymorphic_on": cast(d1, String)
+ ", "
+ cast(func.coalesce(d2, ""), String),
}
@event.listens_for(AbstractQuestion, "init", propagate=True)
def _setup_poly(target, args, kw):
"""receive new AbstractQuestion objects when they are constructed and
set polymorphic identity"""
# this is the ident_() object
ident = inspect(target).mapper.polymorphic_identity
d1, d2 = ident._as_tuple()
kw["d1"] = d1
if d2:
kw["d2"] = d2
class MultiChoiceQuestion(AbstractQuestion):
__tablename__ = "mcq"
id: Column = Column(
Integer, ForeignKey(AbstractQuestion.id), primary_key=True
)
__mapper_args__ = {"polymorphic_identity": ident_(1)}
class AbstractSurveyQuestion(AbstractQuestion):
__tablename__ = "acq"
id: Column = Column(
Integer, ForeignKey(AbstractQuestion.id), primary_key=True
)
__mapper_args__ = {"polymorphic_identity": ident_(2)}
class RatingQuestion(AbstractSurveyQuestion):
__tablename__ = "rq"
id: Column = Column(
Integer, ForeignKey(AbstractSurveyQuestion.id), primary_key=True
)
__mapper_args__ = {"polymorphic_identity": ident_(2, 1)}
e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)
s = Session(e)
s.add(MultiChoiceQuestion())
s.add(RatingQuestion())
s.commit()
s.close()
for q in s.query(AbstractQuestion):
print(q)
then there's the one that maintains your schema fully, a query here looks like:
SELECT aq.id AS aq_id, aq.d1 AS aq_d1, CAST(aq.d1 AS VARCHAR) || ? || CAST(coalesce(acq.d2, ?) AS VARCHAR) AS _sa_polymorphic_on, acq.id AS acq_id, acq.d2 AS acq_d2
FROM aq LEFT OUTER JOIN acq ON aq.id = acq.id
from typing import Tuple, Optional
from sqlalchemy import cast
from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import event
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import inspect
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import Session
Base = declarative_base()
class ident_(str):
"""describe a composed identity.
Using a string for easy conversion to a string SQL composition.
"""
_tup: Tuple[int, Optional[int]]
def __new__(cls, d1, d2=None):
self = super().__new__(cls, f"{d1}, {d2 or ''}")
self._tup = d1, d2
return self
def _as_tuple(self):
return self._tup
class AbstractQuestion(Base):
__tablename__ = "aq"
id = Column(Integer, primary_key=True)
d1 = Column(
Integer, nullable=False
) # this can be your FK to the other table etc.
__mapper_args__ = {
"polymorphic_identity": ident_(0),
}
@event.listens_for(AbstractQuestion, "init", propagate=True)
def _setup_poly(target, args, kw):
"""receive new AbstractQuestion objects when they are constructed and
set polymorphic identity"""
# this is the ident_() object
ident = inspect(target).mapper.polymorphic_identity
d1, d2 = ident._as_tuple()
kw["d1"] = d1
if d2:
kw["d2"] = d2
class MultiChoiceQuestion(AbstractQuestion):
__tablename__ = "mcq"
id: Column = Column(
Integer, ForeignKey(AbstractQuestion.id), primary_key=True
)
__mapper_args__ = {"polymorphic_identity": ident_(1)}
class AbstractSurveyQuestion(AbstractQuestion):
__tablename__ = "acq"
id: Column = Column(
Integer, ForeignKey(AbstractQuestion.id), primary_key=True
)
d2 = Column(Integer, nullable=False)
__mapper_args__ = {
"polymorphic_identity": ident_(2),
"polymorphic_load": "inline", # adds ASQ to all AQ queries
}
# after ASQ is set up, set the discriminator on the base class
# that includes ASQ column
inspect(AbstractQuestion)._set_polymorphic_on(
cast(AbstractQuestion.d1, String)
+ ", "
+ cast(func.coalesce(AbstractSurveyQuestion.d2, ""), String)
)
class RatingQuestion(AbstractSurveyQuestion):
__tablename__ = "rq"
id: Column = Column(
Integer, ForeignKey(AbstractSurveyQuestion.id), primary_key=True
)
__mapper_args__ = {"polymorphic_identity": ident_(2, 1)}
e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)
s = Session(e)
s.add(MultiChoiceQuestion())
s.add(RatingQuestion())
s.commit()
s.close()
for q in s.query(AbstractQuestion):
print(q)
我有以下 table 结构（我尽可能地简化了它，缩小了 child/inheriting table 的范围 [还有其他的] 并从中删除了所有不相关的列提供的 tables):
## Base is my declarative_base
class AbstractQuestion(Base):
questionTypeId: Column = Column(
Integer, ForeignKey("luQuestionTypes.id"), index=True, nullable=False
)
__mapper_args__ = {
"polymorphic_identity": 0,
"polymorphic_on": questionTypeId,
}
class MultiChoiceQuestion(AbstractQuestion):
id: Column = Column(Integer, ForeignKey(AbstractQuestion.id), primary_key=True)
__mapper_args__ = {"polymorphic_identity": 1}
class AbstractSurveyQuestion(AbstractQuestion):
id: Column = Column(Integer, ForeignKey(AbstractQuestion.id), primary_key=True)
surveyQuestionTypeId: Column = Column(
Integer, ForeignKey("luSurveyQuestionTypes.id"), index=True, nullable=False
)
__mapper_args__ = {"polymorphic_identity": 2}
class RatingQuestion(AbstractSurveyQuestion):
id: Column = Column(
Integer, ForeignKey(AbstractSurveyQuestion.id), primary_key=True
)
我面临的挑战是，我正在尝试使 AbstractSurveyQuestion 具有两种类型的多态映射 - 一种作为 AbstractQuestion 的子级，具有 polymorphic_identity匹配 questionTypeId，但我还需要它有一个单独的 polymorphic_on 映射器用于它自己的子 table，即 RatingQuestion。
我能找到的最接近的东西是
谁能帮我解决这个问题？
谢谢！
我在 SQLAlchemy 的 GitHub 存储库上发布了同样的问题。从维护者那里得到了这个答案： https://github.com/sqlalchemy/sqlalchemy/discussions/8089#discussioncomment-2878725
下面的内容我也贴一下:
it sounds like you are looking for mult-level polymorphic_on. We don't support that right now without workarounds, and that's #2555 which is a feature we're unlikely to implement, or if we did it would be a long time from now.
It looks like you are using joined inheritance....so...two ways. The more SQL efficient one is to have an extra "supplemetary" column on your base table that can discriminate for AbstractSurveyQuestion...because if you query for all the AbstractQuestion objects, by default it's just going to query that one table, and needs to know from each row if that row is in fact a RatingQuestion.
the more convoluted way is to use mapper-configured with_polymorphic so that all queries for AbstractQuestion include all the tables (or a subset of tables, can be configured, but at minimum you'd need to join out to AbstractSurveyQuestion) using a LEFT OUTER JOIN (or if you really wanted to go crazy it can be a UNION ALL).
the workarounds are a little ugly since it's not very easy to get a "composed" value out of two columns in SQL, but they are contained to the base classes. Below examples work on SQLite and might need tweaking for other databases.
Here's the discriminator on base table demo, a query here looks like:
SELECT aq.id AS aq_id, aq.d1 AS aq_d1, aq.d2 AS aq_d2, CAST(aq.d1 AS VARCHAR) || ? || CAST(coalesce(aq.d2, ?) AS VARCHAR) AS _sa_polymorphic_on FROM aqfrom typing import Tuple, Optional from sqlalchemy import cast from sqlalchemy import Column from sqlalchemy import create_engine from sqlalchemy import event from sqlalchemy import ForeignKey from sqlalchemy import inspect from sqlalchemy import Integer, func from sqlalchemy import String from sqlalchemy.orm import declarative_base from sqlalchemy.orm import Session Base = declarative_base() class ident_(str): """describe a composed identity. Using a string for easy conversion to a string SQL composition. """ _tup: Tuple[int, Optional[int]] def __new__(cls, d1, d2=None): self = super().__new__(cls, f"{d1}, {d2 or ''}") self._tup = d1, d2 return self def _as_tuple(self): return self._tup class AbstractQuestion(Base): __tablename__ = "aq" id = Column(Integer, primary_key=True) d1 = Column( Integer, nullable=False ) # this can be your FK to the other table etc. d2 = Column( Integer, nullable=True ) # this is a "supplementary" discrim column __mapper_args__ = { "polymorphic_identity": ident_(0), "polymorphic_on": cast(d1, String) + ", " + cast(func.coalesce(d2, ""), String), } @event.listens_for(AbstractQuestion, "init", propagate=True) def _setup_poly(target, args, kw): """receive new AbstractQuestion objects when they are constructed and set polymorphic identity""" # this is the ident_() object ident = inspect(target).mapper.polymorphic_identity d1, d2 = ident._as_tuple() kw["d1"] = d1 if d2: kw["d2"] = d2 class MultiChoiceQuestion(AbstractQuestion): __tablename__ = "mcq" id: Column = Column( Integer, ForeignKey(AbstractQuestion.id), primary_key=True ) __mapper_args__ = {"polymorphic_identity": ident_(1)} class AbstractSurveyQuestion(AbstractQuestion): __tablename__ = "acq" id: Column = Column( Integer, ForeignKey(AbstractQuestion.id), primary_key=True ) __mapper_args__ = {"polymorphic_identity": ident_(2)} class RatingQuestion(AbstractSurveyQuestion): __tablename__ = "rq" id: Column = Column( Integer, ForeignKey(AbstractSurveyQuestion.id), primary_key=True ) __mapper_args__ = {"polymorphic_identity": ident_(2, 1)} e = create_engine("sqlite://", echo=True) Base.metadata.create_all(e) s = Session(e) s.add(MultiChoiceQuestion()) s.add(RatingQuestion()) s.commit() s.close() for q in s.query(AbstractQuestion): print(q)then there's the one that maintains your schema fully, a query here looks like:
SELECT aq.id AS aq_id, aq.d1 AS aq_d1, CAST(aq.d1 AS VARCHAR) || ? || CAST(coalesce(acq.d2, ?) AS VARCHAR) AS _sa_polymorphic_on, acq.id AS acq_id, acq.d2 AS acq_d2 FROM aq LEFT OUTER JOIN acq ON aq.id = acq.idfrom typing import Tuple, Optional from sqlalchemy import cast from sqlalchemy import Column from sqlalchemy import create_engine from sqlalchemy import event from sqlalchemy import ForeignKey from sqlalchemy import func from sqlalchemy import inspect from sqlalchemy import Integer from sqlalchemy import String from sqlalchemy.orm import declarative_base from sqlalchemy.orm import Session Base = declarative_base() class ident_(str): """describe a composed identity. Using a string for easy conversion to a string SQL composition. """ _tup: Tuple[int, Optional[int]] def __new__(cls, d1, d2=None): self = super().__new__(cls, f"{d1}, {d2 or ''}") self._tup = d1, d2 return self def _as_tuple(self): return self._tup class AbstractQuestion(Base): __tablename__ = "aq" id = Column(Integer, primary_key=True) d1 = Column( Integer, nullable=False ) # this can be your FK to the other table etc. __mapper_args__ = { "polymorphic_identity": ident_(0), } @event.listens_for(AbstractQuestion, "init", propagate=True) def _setup_poly(target, args, kw): """receive new AbstractQuestion objects when they are constructed and set polymorphic identity""" # this is the ident_() object ident = inspect(target).mapper.polymorphic_identity d1, d2 = ident._as_tuple() kw["d1"] = d1 if d2: kw["d2"] = d2 class MultiChoiceQuestion(AbstractQuestion): __tablename__ = "mcq" id: Column = Column( Integer, ForeignKey(AbstractQuestion.id), primary_key=True ) __mapper_args__ = {"polymorphic_identity": ident_(1)} class AbstractSurveyQuestion(AbstractQuestion): __tablename__ = "acq" id: Column = Column( Integer, ForeignKey(AbstractQuestion.id), primary_key=True ) d2 = Column(Integer, nullable=False) __mapper_args__ = { "polymorphic_identity": ident_(2), "polymorphic_load": "inline", # adds ASQ to all AQ queries } # after ASQ is set up, set the discriminator on the base class # that includes ASQ column inspect(AbstractQuestion)._set_polymorphic_on( cast(AbstractQuestion.d1, String) + ", " + cast(func.coalesce(AbstractSurveyQuestion.d2, ""), String) ) class RatingQuestion(AbstractSurveyQuestion): __tablename__ = "rq" id: Column = Column( Integer, ForeignKey(AbstractSurveyQuestion.id), primary_key=True ) __mapper_args__ = {"polymorphic_identity": ident_(2, 1)} e = create_engine("sqlite://", echo=True) Base.metadata.create_all(e) s = Session(e) s.add(MultiChoiceQuestion()) s.add(RatingQuestion()) s.commit() s.close() for q in s.query(AbstractQuestion): print(q)