Clone wiki

sqlalchemy / UsageRecipes / RelationshipOnCast


Here we illustrate the techniques currently required to construct a "primaryjoin" where one side needs to be run through the CAST function. An extra directive, _local_remote_pairs, gives relationship() the clues it needs to set itself up. A special coercing type which coerces strings to ints provides type safe behavior in the case that we'd like the relationship to work on the persistence/flush side.

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class StringAsInt(TypeDecorator):
    """Coerce string->integer type.

    This is needed only if the relationship() from
    int to string is writable, as SQLAlchemy will copy
    the string parent values into the integer attribute
    on the child during a flush.

    impl = Integer
    def process_bind_param(self, value, dialect):
        if value is not None:
            value = int(value)
        return value

class A(Base):
    """Parent.   The referenced column is a string type."""
    __tablename__ = 'a'
    id = Column(Integer, primary_key=True)
    a_id = Column(String)

class B(Base):
    """Child.  The column we reference 'A' with is an integer."""

    __tablename__ = 'b'
    id = Column(Integer, primary_key=True)
    a_id = Column(StringAsInt)

# specify primaryjoin using CAST.  Currently, SQLAlchemy needs
# a little help identifying the correct entities 
# from the primary join - the semi-private _local_remote_pairs
# argument can be used to give it the push it needs.
B.a = relationship(A, 
            primaryjoin=cast(A.a_id, Integer)==B.a_id, 
            _local_remote_pairs=[(B.__table__.c.a_id, A.__table__.c.a_id)],

            # _local_remote_pairs doesn't automatically propagate to 
            # the backref, so need it here as well.
                            (A.__table__.c.a_id, B.__table__.c.a_id)

# we demonstrate with SQLite, but the important part
# is the CAST rendered in the SQL output.

e = create_engine('sqlite://', echo=True)

s = Session(e)

    A(a_id="2", bs=[B(), B()]),
    A(a_id="3", bs=[B()]),

b1 = s.query(B).filter_by(a_id="2").first()
print b1.a

a1 = s.query(A).filter_by(a_id="2").first()