如何使用 Flask SQLAlchemy 会话使用 SQLAlchemy 中的“join”返回查询以将其用作 Flask 结果？

Question

我有以下情况：

我想创建一个路由，使用 SQLAlchemy 中的

models

函数，返回

join

（表）上的列具有相同值的所有行。

我在服务内部创建的方法是这样的，其中也包含了可以运行的纯SQL，供参考：

@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

这些是我正在使用的

to_dict

方法的模型：

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
        }

当我使用上述方法执行查询时，

fetch_sectors()

，我得到的是以下错误，来自

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

我启用了 SQLAlchemy 的日志记录，使用

logging

表示

sqlalchemy.engine

，这是 SQLAlchemy 生成的 SELECT：

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]

我现在正在尝试解决这个问题几个小时，在 StackOverflow 上搜索了多个问题，与 GPT 4.0 讨论了几个小时，阅读了 Flask SQLAlchemy 和 SQLAlchemy 文档，但我感觉我在这里陷入了解决方案的死胡同......

这是 StackOverflow 上的一个问题，几乎解决了我的问题，但我无法实现它并在一个小时左右后放弃：使用 Flask-SqlAlchemy 查询关系数据库

请问有人可以帮助我吗？

Answer 1

staticmethod

将方法分配给 class，而不是该类的实例。所以

secort.sectors_to_dict

中的

return [sector.sectors_to_dict() for sector in query.all()], None

应该是

Sectors.sectors_to_dict(sector)

。