How to return a query using `join` from SQLAlchemy to use it as a Flask result, using Flask SQLAlchemy Session?

Question

I have the following situation:

I want to create a route that will return all rows that a column on both models (Tables) have the same values, using join function from SQLAlchemy.

The method I created inside my service is like this, which also contains the pure SQL that works, for reference:

@staticmethod
def fetch_sectors():
    """
    Fetches all sectors with their respective branches, using an inner join with the equipment table.
    As this is a JOIN involving more than one entity, and not a direct query in the model, it's necessary
    to use the session.query() from SQLAlchemy - https://docs.sqlalchemy.org/en/14/orm/query.html

    The Controller Object responsible for this is db_sql.session, instantiated by Flask's SQLAlchemy.

    Conversion of the below flow to SQL for reference - INNER JOIN + DISTINCT:
    -- JOIN: Show Columns/Records in Common between tables.
    SELECT DISTINCT EQUIPMENT.T9_BRANCH AS sector_branch,
                    SECTORS.T6_CODE AS sector_code,
                    SECTORS.T6_NAME AS sector_name
        FROM ST9010 AS EQUIPMENT
        JOIN ST6010 AS SECTORS ON SECTORS.T6_CODE = EQUIPMENT.T9_CODE;

    :return: All Sectors
    """
    print("Creating the Query, with INNER JOIN + DISTINCT.")
    query = db_sql.session.query(
        Equipment.equipment_branch.label('sector_branch'),
        Sectors.sector_code,
        Sectors.sector_name
    ).join(
        Sectors, Sectors.sector_code == Equipment.equipment_sector
    ).distinct()

    print("Returning the Sectors.")
    return [sector.sectors_to_dict() for sector in query.all()], None

Those are the models with the to_dict methods I am using:

class Equipment(db_sql.Model):
    __tablename__ = 'ST9010'  # Assets Table - Protheus

    equipment_id: Mapped[int] = mapped_column("T9_EQUIPID", db_sql.Integer, primary_key=True)
    equipment_branch: Mapped[str] = mapped_column("T9_BRANCH", db_sql.String, primary_key=True)
    equipment_sector: Mapped[str] = mapped_column("T9_CODE", db_sql.String, primary_key=True)
    equipment_name: Mapped[str] = mapped_column("T9_NAME", db_sql.String, nullable=False)
    equipment_costcenter: Mapped[str] = mapped_column("T9_COSTCENTER", db_sql.String, nullable=False)
    DELETED: Mapped[str] = mapped_column(db_sql.String, nullable=True)
    T9_STATUS: Mapped[str] = mapped_column(db_sql.String, nullable=True)

    def to_dict(self):
        return {
            "equipment_id": self.equipment_id,
            "equipment_branch": self.equipment_branch,
            "equipment_sector": self.equipment_sector,
            "equipment_name": self.equipment_name,
            "equipment_costcenter": self.equipment_costcenter
        }


class Sectors(db_sql.Model):
    __tablename__ = 'ST6010'  # Families Table - Protheus

    # T6_BRANCH blank: In the Query, do an Inner Join with T9_BRANCH of ST9010
    sector_branch = mapped_column("T6_BRANCH", db_sql.String, primary_key=True)
    sector_code = mapped_column("T6_CODE", db_sql.String, primary_key=True)
    sector_name = mapped_column("T6_NAME", db_sql.String, nullable=False)
    DELETED = mapped_column(db_sql.String, nullable=True)

    def to_dict(self):
        return {
            "sector_branch": self.sector_branch,
            "sector_code": self.sector_code,
            "sector_name": self.sector_name
        }

    @staticmethod
    def sectors_to_dict(result):
        return {
            "sector_branch": result.sector_branch,
            "sector_code": result.sector_code,
            "sector_name": result.sector_name,
            "equipment_branch": result.equipment_branch
        }

When I executed the query using the above method, fetch_sectors(), what I got was the following error, from SQLAlchemy Engine:

_key_fallback
    raise KeyError(key) from err
KeyError: 'sectors_to_dict'

_key_not_found
    raise AttributeError(ke.args[0]) from ke
AttributeError: sectors_to_dict

I enabled logging for SQLAlchemy, using logging for sqlalchemy.engine, and this is the SELECT that SQLAlchemy Generated:

INFO:sqlalchemy.engine.Engine:SELECT DISTINCT [ST9010].[T9_BRANCH] AS sector_branch, [ST6010].[T6_CODE] AS [ST6010_T6_CODE], [ST6010].[T6_NAME] AS [ST6010_T6_NAME]
FROM [ST9010] JOIN [ST6010] ON [ST6010].[T6_CODE] = [ST9010].[T9_CODE]

I am tryting to fix this for hours now, searched multiple questions here on StackOverflow, talked with GPT 4.0 for hours, read Flask SQLAlchemy and SQLAlchemy documentation, but am feeling that I am getting to a dead end of solutions here...

This is a question here at StackOverflow that almost solved my problem, but I could'nt implement it and gave up after an hour or so: Use Flask-SqlAlchemy to query relationship database

Can please, someone help me?

Answer 1

to_dict is a method on your models classes, Sectors and Equipment, so given an instance of either of these classes you can call instance.to_dict() and have a dictionary returned. However you query does not return instances, because you are querying for individual attributes

session.query(
        Equipment.equipment_branch.label('sector_branch'),
        Sectors.sector_code,
        Sectors.sector_name
    )

not model classes (like session.query(Equipment). This kind of query returns a Row objects, not model instances.

Happily, the query result has a mappings method that will convert the rows to mappings, so you can change your return statement to this:

return list(query.mappings()), None

If you require actual dicts (perhaps you intend to serialise them to JSON?) you can return like this:

return [dict(m) for m in query.mappings()], None