我有以下情况:
我想创建一条路由,它将使用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
有人能帮帮我吗?