SQLAlchemy declarative property from join (single attribute, not whole object)

Question

I wish to create a mapped attribute of an object which is populated from another table.

Using the SQLAlchemy documentation example, I wish to make a user_name field exist on the Address class such that it can be both easily queried and easily accessed (without a second round trip to the database)

For example, I wish to be able to query and filter by user_name Address.query.filter(Address.user_name == 'wcdolphin').first() And also access the user_name attribute of all Address objects, without performance penalty, and have it properly persist writes as would be expected of an attribute in the __tablename__

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    addresses = relation("Address", backref="user")

class Address(Base):
    __tablename__ = 'addresses'

    id = Column(Integer, primary_key=True)
    email = Column(String(50))
    user_name = Column(Integer, ForeignKey('users.name'))#This line is wrong

How do I do this?

I found the documentation relatively difficult to understand, as it did not seem to conform to most examples, especially the Flask-SQLAlchemy examples.

Answer 1

You can do this with a join on the query object, no need to specify this attribute directly. So your model would look like:

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import sessionmaker, relation
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
engine = create_engine('sqlite:///')
Session = sessionmaker(bind=engine)

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    addresses = relation("Address", backref="user")

class Address(Base):
    __tablename__ = 'addresses'
    id = Column(Integer, primary_key=True)
    email = Column(String(50))
    user_id = Column(Integer, ForeignKey("users.id"))


Base.metadata.create_all(engine)

A query after addresses with filtering the username looks like:

>>> session = Session()
>>> session.add(Address(user=User(name='test')))
>>> session.query(Address).join(User).filter(User.name == 'test').first()
<__main__.Address object at 0x02DB3730>

Edit: As you can directly access the user from an address object, there is no need for directly referencing an attribute to the Address class:

>>> a = session.query(Address).join(User).filter(User.name == 'test').first()
>>> a.user.name
'test'

Answer 2

If you truly want Address to have a SQL enabled version of "User.name" without the need to join explicitly, you need to use a correlated subquery. This will work in all cases but tends to be inefficient on the database side (particularly with MySQL), so there is possibly a performance penalty on the SQL side versus using a regular JOIN. Running some EXPLAIN tests may help to analyze how much of an effect there may be.

Another example of a correlated column_property() is at http://docs.sqlalchemy.org/en/latest/orm/mapped_sql_expr.html#using-column-property.

For the "set" event, a correlated subquery represents a read-only attribute, but an event can be used to intercept changes and apply them to the parent User row. Two approaches to this are presented below, one using regular identity map mechanics, which will incur a load of the User row if not already present, the other which emits a direct UPDATE to the row:

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

Base= declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    addresses = relation("Address", backref="user")

class Address(Base):
    __tablename__ = 'addresses'

    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    email = Column(String(50))

Address.user_name = column_property(select([User.name]).where(User.id==Address.id))

from sqlalchemy import event
@event.listens_for(Address.user_name, "set")
def _set_address_user_name(target, value, oldvalue, initiator):
    # use ORM identity map + flush
    target.user.name = value

    # use direct UPDATE
    #object_session(target).query(User).with_parent(target).update({'name':value})

e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)

s = Session(e)
s.add_all([
    User(name='u1', addresses=[Address(email='e1'), Address(email='e2')])
])
s.commit()

a1 = s.query(Address).filter(Address.user_name=="u1").first()
assert a1.user_name == "u1"

a1.user_name = 'u2'
s.commit()

a1 = s.query(Address).filter(Address.user_name=="u2").first()
assert a1.user_name == "u2"