Relationship Based On Expression

[Bobby Rullo]

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?

Thanks in advance,

Bobby

[Mike Bayer]

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 sqlalchemy+...@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/sqlalchemy/0083c5d4-cd66-4346-8108-f85a8909d69an%40googlegroups.com.

[Bobby Rullo]

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?

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 sqlalchemy+...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/sqlalchemy/491ffccb-bd66-4dc2-a29f-2a73943f8c22%40www.fastmail.com.

[Mike Bayer]

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()

To view this discussion on the web visit https://groups.google.com/d/msgid/sqlalchemy/CAF2wT%3D64rdoc7ZE9SEH9yrm4NLqicqJpZzvyP7jnXjw8OVGPSQ%40mail.gmail.com.

[Bobby Rullo]

Wow - thank you so much - that was really above and beyond. There's a lot to digest there so I need to pore it over, but I think I get the general idea.

Bobby

To view this discussion on the web visit https://groups.google.com/d/msgid/sqlalchemy/fe9301ad-e0cc-4d8c-811d-213e53a48ba9%40www.fastmail.com.