如何在 Fastapi & SqlAlchemy 中连接来自 2 个不同数据库 MSSQL 的 2 个表

Question

我有两个类，如下:

class Model(Base):
__tablename__ = 'Model'
__table_args__ = {'schema': 'Smart_Tooling_Platform_Test.dbo'}
factory_id = Column(String, primary_key=True, index=True)
model_no = Column(String, primary_key=True, index=True)
model_name = Column(String, index=True)
model_type_id = Column(String, index=True)
model_family = Column(String, index=True, default="")
upper_id = Column(String, index=True)
dev_season = Column(String, index=True)
prod_season = Column(String, index=True)
volume = Column(Float, default=None)
volume_percent = Column(Float, default=None)
remarks = Column(String, default="")
model_picture = Column(String)
is_active = Column(Boolean)
top_model = Column(Boolean)
pilot_line = Column(Boolean)
create_by = Column(String)
create_time = Column(DateTime)
update_by = Column(String)
update_time = Column(DateTime)

class Type(Base):
__tablename__ = "Type"
__table_args__ = {'schema': 'ChamCongDungB1.dbo'}
factory_id = Column(String, primary_key=True, index=True)
model_type_id = Column(String, primary_key=True, index=True)
model_type_name = Column(String)
is_active = Column(Boolean)

这两个类位于2个不同的MSSQL数据库中。现在我想写一个查询，根据2列model_type_id和factory_id使用SqlAlchemy在Fastapi连接这2个表。我已经尝试了很多方法，但都没有成功。请大家帮忙。

I try that:

async def get_data_from_2_db(self, model_param):
query = self.session.query(Model, Type).join(Type, and_(Model.model_type_id == Type.model_type_id,
Model.factory_id == Type.factory_id)).all()
return query

但有问题

sqlalchemy.exc.ProgrammingError: (pyodbc.ProgrammingError) ('42S02', "[42S02] [Microsoft][ODBC Driver 17 for SQL Server][SQL Server]無效的物件名稱 'ChamCongDungB1.dbo.Type'。 (208) (SQLExecDirectW)")

Answer 1

您需要使用__table_args属性定义模式来自不同的数据表这篇文章可能会有所帮助。