Is there a way to use attributes on a relationship attribute when selecting models using `where`?

[taranlu-houzz]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the SQLModel documentation, with the integrated search.
• I already searched in Google "How to X in SQLModel" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to SQLModel but to Pydantic.
• I already checked if it is not related to SQLModel but to SQLAlchemy.

Commit to Help

• I commit to help with one of those options 👆

Example Code

import logging
from typing import List

from sqlmodel import (
    Field,
    Relationship,
    SQLModel,
    Session,
    create_engine,
    select,
)


logging.basicConfig(level=logging.INFO)

sqlite_url = "sqlite://"

engine = create_engine(sqlite_url)
# engine = create_engine(sqlite_url, echo=True)


class Product(SQLModel, table=True):

    id: str = Field(primary_key=True)
    product_status: int = 0

    source_status: str = Field(foreign_key="source.source_status")
    source: "Source" = Relationship(back_populates="products")


class Source(SQLModel, table=True):

    id: int = Field(primary_key=True)
    source_status: int = 0
    another_attribute: bool = False

    products: List["Product"] = Relationship(back_populates="source")


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def add_data():

    with Session(engine) as session:
        source_1 = Source(id=1, another_attribute=True)
        source_2 = Source(id=2, source_status=2)

        product_a = Product(id=1, source=source_1)
        product_b = Product(id=2, source=source_1)
        product_c = Product(id=3, source=source_2)
        product_d = Product(id=4, source=source_2)
        product_e = Product(id=5, source=source_2)

        session.add(product_a)
        session.add(product_b)
        session.add(product_c)
        session.add(product_d)
        session.add(product_e)

        session.commit()


def get_data_bad():

    with Session(engine) as session:
        db_statement = select(Product).where(Product.source.another_attribute == False)
        result = session.exec(db_statement).all()

        return result


def get_data_bad2():

    with Session(engine) as session:
        db_statement = select(Product).where(Product.source.source_status == 2)
        result = session.exec(db_statement).all()

        return result


def get_data_good():

    with Session(engine) as session:
        db_statement = select(Product).where(Product.source_status == 2)
        result = session.exec(db_statement).all()

        return result


def main():

    create_db_and_tables()

    add_data()

    try:
        bad_data = get_data_bad()
    except AttributeError as err:
        bad_data = None
        logging.error(err)

    try:
        bad_data2 = get_data_bad2()
    except AttributeError as err:
        bad_data2 = None
        logging.error(err)

    good_data = get_data_good()

    logging.info(f"BAD DATA: {bad_data}")

    logging.info(f"BAD DATA2: {bad_data2}")

    logging.info("GOOD DATA:")
    for item in good_data:
        logging.info(f"{' ' * 4}{repr(item)}")


if __name__ == "__main__":

    main()

Description

I am wondering if it is possible to use the where to select based on attributes from a relationship. The example I posted above produces the following output:

ERROR:root:Neither 'InstrumentedAttribute' object nor 'Comparator' object associated with Product.source has an attribute 'another_attribute'
ERROR:root:Neither 'InstrumentedAttribute' object nor 'Comparator' object associated with Product.source has an attribute 'source_status'
/Users/---/lib/python3.10/site-packages/sqlmodel/orm/session.py:60: SAWarning: Class SelectOfScalar will not make use of SQL compilation caching as it does not set the 'inherit_cache' attribute to ``True``.  This can have significant performance implications including some performance degradations in comparison to prior SQLAlchemy versions.  Set this attribute to True if this object can make use of the cache key generated by the superclass.  Alternatively, this attribute may be set to False which will disable this warning. (Background on this error at: https://sqlalche.me/e/14/cprf)
  results = super().execute(
INFO:root:BAD DATA: None
INFO:root:BAD DATA2: None
INFO:root:GOOD DATA:
INFO:root:    Product(product_status=0, id='3', source_status='2')
INFO:root:    Product(product_status=0, id='4', source_status='2')
INFO:root:    Product(product_status=0, id='5', source_status='2')

Operating System

macOS

Operating System Details

No response

SQLModel Version

0.0.4

Python Version

3.10.2

Additional Context

No response

[byrman]

Yes, there is a way:

db_statement = select(Product).where(Product.source.has(another_attribute = False))

But if you inspect the generated SQL, you'll see that you are probably better off using a join.

[taranlu-houzz]

@byrman I see, thanks for the help. Just confirming: this is not in the SQLModel documentation currently, right? Is that because it is actually a SQLAlchemy feature?

I ended up using multiple foreign keys to get the result I wanted (which does appear to create joins). I'm don't really have a lot of db experience, but I ended up using something like this:

Updated Example

import logging
from typing import List

from sqlmodel import (
    Field,
    Relationship,
    SQLModel,
    Session,
    create_engine,
    select,
)


logging.basicConfig(level=logging.INFO)

sqlite_url = "sqlite://"

# engine = create_engine(sqlite_url)
engine = create_engine(sqlite_url, echo=True)


class Product(SQLModel, table=True):

    id: str = Field(primary_key=True)
    product_status: int = 0

    source_status: str = Field(foreign_key="source.source_status")
    another_attribute: bool = Field(foreign_key="source.another_attribute")

    source: "Source" = Relationship(
        back_populates="products",
        sa_relationship_kwargs={
            "primaryjoin": "Product.source_status==Source.source_status",
            "lazy": "joined",
        },
    )


class Source(SQLModel, table=True):

    id: int = Field(primary_key=True)
    source_status: int = 0
    another_attribute: bool = False

    products: List["Product"] = Relationship(
        back_populates="source",
        sa_relationship_kwargs={
            "primaryjoin": "Source.source_status==Product.source_status",
            "lazy": "joined",
        },
    )


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def add_data():

    with Session(engine) as session:
        source_1 = Source(id=1, another_attribute=True)
        source_2 = Source(id=2, source_status=2)

        product_a = Product(
            id=1,
            source_status=source_1.source_status,
            another_attribute=source_1.another_attribute,
            source=source_1,
        )
        product_b = Product(
            id=2,
            source_status=source_1.source_status,
            another_attribute=source_1.another_attribute,
            source=source_1,
        )
        product_c = Product(
            id=3,
            source_status=source_2.source_status,
            another_attribute=source_2.another_attribute,
            source=source_2,
        )
        product_d = Product(
            id=4,
            source_status=source_2.source_status,
            another_attribute=source_2.another_attribute,
            source=source_2,
        )
        product_e = Product(
            id=5,
            source_status=source_2.source_status,
            another_attribute=source_2.another_attribute,
            source=source_2,
        )

        session.add(product_a)
        session.add(product_b)
        session.add(product_c)
        session.add(product_d)
        session.add(product_e)

        session.commit()


def get_data_good3():

    with Session(engine) as session:
        db_statement = select(Product).where(Product.another_attribute == True)
        result = session.exec(db_statement).unique().all()

        return result


def main():

    create_db_and_tables()

    add_data()

    good_data3 = get_data_good3()

    logging.info("GOOD DATA3:")
    for item in good_data3:
        logging.info(f"{' ' * 4}{repr(item)}")


if __name__ == "__main__":

    main()

Not really sure if this is a good solution, but it seems to work.

[byrman]

Is that because it is actually a SQLAlchemy feature?

I guess so. The SQLAlchemy is extensive, there is no point in copying all that.

Not really sure if this is a good solution, but it seems to work.

You may read more abut where / join here: https://sqlmodel.tiangolo.com/tutorial/connect/read-connected-data/

[taranlu-houzz]

So, I actually ended up using @byrman's suggestion because I was running into other issues with my multiple foreign keys approach. Just to clarify, in case someone sees this in the future, the suggestion to use Product.source.has(another_attribute=False) above is specific to my example. The name source is not special (this confused me for a moment), and your editor may not be able to give a completion for has (in VSCode, I did not get a completion for it).