我有以下情况:

我想创建一条路由,它将使用SQLAlChemy中的join函数返回两个models(表)上的列具有相同值的所有行.

我在我的服务中创建的方法如下所示,它还包含可以工作的纯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启用了日志(log)记录,对sqlalchemy.engine使用logging,这是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交谈了几个小时,阅读了FlaskSQLAlChemy和SQLAlChemy文档,但我觉得我在这里走进了解决方案的死胡同……

这是StackOverflow的一个问题,几乎解决了我的问题,但我无法实现它,大约一个小时后就放弃了:Use Flask-SqlAlchemy to query relationship database

有人能帮帮我吗?

推荐答案

to_dict是您的模型类SectorsEquipment上的一个方法,因此给定这两个类的一个实例,您可以调用instance.to_dict()并返回一个字典.但是,您要查询does not return instances,因为您要查询的是单个属性

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

而不是模型类(如session.query(Equipment). 这种查询返回Row对象,而不是模型实例.

幸运的是,查询result有一个将rows转换为mappingsmappings方法,因此您可以将返回语句更改为:

return list(query.mappings()), None

如果您需要实际的dict(也许您打算将它们序列化为JSON?),您可以这样返回:

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

Python相关问答推荐

根据多列和一些条件创建新列

每个组每第n行就有Pandas

获取Azure Pipelines以从pyproject.toml(而不是relevments_dev.文本)安装测试环境

Tkinter -控制调色板的位置

模型序列化器中未调用现场验证器

使用多个性能指标执行循环特征消除

ambda将时间戳与组内另一列的所有时间戳进行比较

opencv Python稳定的图标识别

Python中是否有方法从公共域检索搜索结果

如何根据另一列值用字典中的值替换列值

try 在树叶 map 上应用覆盖磁贴

类型错误:输入类型不支持ufuncisnan-在执行Mann-Whitney U测试时[SOLVED]

将数据框架与导入的Excel文件一起使用

在Python Attrs包中,如何在field_Transformer函数中添加字段?

如何使用数组的最小条目拆分数组

python中字符串的条件替换

在Python中调用变量(特别是Tkinter)

基于形状而非距离的两个numpy数组相似性

下三角形掩码与seaborn clustermap bug

合并与拼接并举