OK what you're trying to do is a little hard , and yes declare_last /
declare_first are useful here, because I just noticed you need to inspect the
PK of the local class, not the remote one, so that has to be set up first. So
here is a demo based on declare_first, this is the basic idea, either with the
FK constraint or with a primary join condition:
from __future__ import annotations
from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import ForeignKeyConstraint
from sqlalchemy import inspect
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import configure_mappers
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
Base = declarative_base()
class TransitionBase(Base):
__abstract__ = True
@declared_attr
def id(cls):
return Column(Integer, primary_key=True)
state = Column(String, nullable=False, index=True)
class HasStateMachineMixin:
@staticmethod
def get_state_class() -> Type[TransitionBase]:
raise NotImplementedError()
@classmethod
def __declare_first__(cls):
dest = cls.get_state_class()
src = inspect(cls)
dest_cols = [
Column("%s_%s" % (src.local_table.name, pk.name), pk.type)
for pk in src.primary_key
]
# make a ForeignKeyConstraint. if you wanted to just make a
# primaryjoin, you could create it
# primaryjoin=and_(
# *[(a==foreign(b)) for a, b in zip(src.primary_key, dest_cols)])
dest.__table__.append_constraint(
ForeignKeyConstraint(dest_cols, src.primary_key)
)
# these two steps make use of the DeclarativeMeta to receive
# new columns and attributes on the fly
for dc in dest_cols:
setattr(dest, "_%s" % dc.name, dc)
cls.transitions = relationship(dest, order_by=dest.id.desc())
class Transition(TransitionBase):
__tablename__ = "transitions"
id = Column(Integer, primary_key=True)
class Obj(HasStateMachineMixin, Base):
__tablename__ = "obj"
id = Column(String, primary_key=True)
@staticmethod
def get_state_class() -> Type[TransitionBase]:
return Transition
class ThreePrimaryKeys(HasStateMachineMixin, Base):
__tablename__ = "three_pks"
a = Column(String, primary_key=True)
b = Column(String, primary_key=True)
c = Column(String, primary_key=True)
@staticmethod
def get_state_class() -> Type[TransitionBase]:
return Transition
# since the mappers are going to add new columns, we need to make
# sure mapper configure is triggered before we render the DDL. this
# ensures the declare_first above runs.
configure_mappers()
e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)
s = Session(e)
s.add(
ThreePrimaryKeys(
a="a",
b="b",
c="c",
transitions=[
Transition(state="t1"),
Transition(state="t2"),
Transition(state="t3"),
],
)
)
s.add(Obj(id="one", transitions=[Transition(state="tt1")]))
s.commit()
On Wed, Sep 9, 2020, at 7:44 PM, Bobby Rullo wrote:
> Thanks for the reply Mike!
>
> I tried to go down the "dynamically add multiple obj_ids" but I could not
> figure it out. The obvious choice for dynamic stuff is @declared_attr but
> that only let's me define one thing. How would I do *n* things?
>
> Is this a situation where __declare_last__ could help?
>
>
> On Wed, Sep 9, 2020 at 3:55 PM Mike Bayer <[email protected]> wrote:
>> __
>>
>>
>> On Wed, Sep 9, 2020, at 2:36 PM, Bobby Rullo wrote:
>>> Hi there,
>>>
>>> I'm trying to create a relationship for a Mxin that is agnostic to what the
>>> primary key of the mixed object is.
>>>
>>> Basically we have this:
>>>
>>> class TransitionBase(SurrogatePK, Model):
>>> __abstract__ = True
>>>
>>> obj_id = Column(String, nullable=False, index=True)
>>> state = Column(String, nullable=False, index=True)
>>> ...(more stuff)...
>>>
>>> class HasStateMixin:
>>> @staticmethod
>>> def get_state_class() -> Type[TransitionBase]:
>>> raise NotImplementedError()
>>>
>>> @declared_attr
>>> def transitions(cls) -> List[TransitionBase]:
>>> state_cls = cls.get_state_class()
>>> return relationship(
>>> state_cls,
>>> primaryjoin=foreign(state_cls.obj_id) == remote(cls.id), #
>>> type: ignore
>>> order_by=state_cls.id.desc(),
>>> )
>>>
>>> And it works well, as long as the `id` property is a normal Column. Eg:
>>>
>>> class Transition(TransitionBase):
>>> __tablename__ = ''transitions'
>>>
>>>
>>> class Obj(Model, HasStateMachineMixin):
>>> __tablename__ = 'obj'
>>> __table_args__ = {'schema': test_schema}
>>>
>>> id = Column(String, primary_key=True)
>>>
>>> @staticmethod
>>> def get_state_class() -> Type[TransitionBase]:
>>> return Transition
>>>
>>> But now I have a case where the primary key of the mixed class is three
>>> columns. My first instinct was to use a hybrid property like this:
>>>
>>> class ThreePrimaryKeys(HasStateMachineMixin):
>>> a = Column(String, primary_key=True)
>>> b = Column(String, primary_key=True)
>>> c = Column(String, primary_key=True)
>>>
>>> @hybrid_property
>>> def id(self):
>>> return f'/{self.a}/{self.b}/{self.c}'
>>>
>>> @id.expression
>>> def id(cls):
>>> return func.concat(
>>> '/', cls.a, '/', cls.b, '/', cls.c,
>>> )
>>>
>>> But that doesn't work: when I create a ThreePrimaryKeys() and then call
>>> obj.transitions.append(MyTransition(state='foo')) it doesn't properly
>>> persist the full concatenated ID - weirdly, it just uses column 'c' (or the
>>> equivalent - type names have been changed to protect the innocent)
>>>
>>> My question is: should this work? And if so, what am I doing wrong? If not,
>>> do you have an alternate approach?
>>
>> I'm not really sure why it doesn't work but the correct approach here is to
>> have a composite ForeignKeyConstraint on the class that refers to
>> ThreePrimaryKeys, or at least a join condition that links the three columns
>> separately to three columns on the referencing object (but there should
>> really be a FK constraint, as it looks like you are trying to do reasonable
>> relational schema design). you can get information on what kind of primary
>> key a mapped class has by using inspect(class).primary_key . your
>> TransitionBase thing would need to dynamically have multiple "obj_id"
>> columns added based on what kind of target it is hitting.
>>
>> There's a lot of ways to do this, including in your transitions() method,
>> you'd need to look at the class you're linking to, get the list of primary
>> key columns, then add that many FK columns to the immediate class, set up a
>> ForeignKeyConstraint for them and add that to the Table also (like
>> cls.__table__.append_constraint()) , then the relationship() would just
>> work. or you could build up the primary join using and_(c1 == fk1, c2 ==
>> fk2, ..)
>>
>> a bit of a handwavy answer but that's the general idea. making a
>> concatenated string like that is not good relational design, it's
>> denormalized and can't be properly indexed.
>>
>>
>>
>>>
>>> Thanks in advance,
>>>
>>> Bobby
>>>
>>>
>>> --
>>> 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 view this discussion on the web visit
>>> https://groups.google.com/d/msgid/sqlalchemy/0083c5d4-cd66-4346-8108-f85a8909d69an%40googlegroups.com
>>>
>>> <https://groups.google.com/d/msgid/sqlalchemy/0083c5d4-cd66-4346-8108-f85a8909d69an%40googlegroups.com?utm_medium=email&utm_source=footer>.
>>
>>
>> --
>> 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 a topic in the
>> Google Groups "sqlalchemy" group.
>> To unsubscribe from this topic, visit
>> https://groups.google.com/d/topic/sqlalchemy/mJjkaXNxp4Y/unsubscribe.
>> To unsubscribe from this group and all its topics, send an email to
>> [email protected].
>> To view this discussion on the web visit
>> https://groups.google.com/d/msgid/sqlalchemy/491ffccb-bd66-4dc2-a29f-2a73943f8c22%40www.fastmail.com
>>
>> <https://groups.google.com/d/msgid/sqlalchemy/491ffccb-bd66-4dc2-a29f-2a73943f8c22%40www.fastmail.com?utm_medium=email&utm_source=footer>.
>
> --
> 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 view this discussion on the web visit
> https://groups.google.com/d/msgid/sqlalchemy/CAF2wT%3D64rdoc7ZE9SEH9yrm4NLqicqJpZzvyP7jnXjw8OVGPSQ%40mail.gmail.com
>
> <https://groups.google.com/d/msgid/sqlalchemy/CAF2wT%3D64rdoc7ZE9SEH9yrm4NLqicqJpZzvyP7jnXjw8OVGPSQ%40mail.gmail.com?utm_medium=email&utm_source=footer>.
--
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 view this discussion on the web visit
https://groups.google.com/d/msgid/sqlalchemy/fe9301ad-e0cc-4d8c-811d-213e53a48ba9%40www.fastmail.com.
