如何将指定SQL查询转换为SQLAlchemy代码？已有外键关联表

原始SQL查询 #

select listaproduccion.id, estado.name, orquestador.name, comentario.name, tipo.name,responsable.name, compania.name
from listaproduccion 
inner join estado on listaproduccion.estado_id=estado.id
inner join orquestador on listaproduccion.orquestador_id=orquestador.id
inner join comentario on listaproduccion.comentario_id=comentario.id
inner join compania on listaproduccion.compania_id=compania.id
inner join tipo on listaproduccion.tipo_id=tipo.id
inner join responsable  on listaproduccion.responsable_id=responsable.id;

尝试的Python代码 #

def get_list(db: Session):
    return db.query(
        model.ListaProduccion,
        model.Estado,
        model.Compania,
        model.Orquestador,
        model.Tipo,
        model.Responsable,
        model.Comentario
        ).join(ListaProduccion, ListaProduccion.estado_id, 
        ListaProduccion.compania_id,
        ListaProduccion.orquestador_id,
        ListaProduccion.tipo_id,
        ListaProduccion.responsable_id,
        ListaProduccion.comentario_id
        ).filter(str(ListaProduccion.estado_id) == Estado.name
        ).filter(str(ListaProduccion.compania_id) == Compania.name
        ).filter(str(ListaProduccion.orquestador_id) == Orquestador.name
        ).filter(str(ListaProduccion.tipo_id) == Tipo.name
        ).filter(str(ListaProduccion.responsable_id) == Responsable.name
        ).filter(str(ListaProduccion.comentario_id) == Comentario.name
        ).all()

报错信息 #

• TypeError: join() 方法仅接受2-4个位置参数，但传入了7个
• 此外代码中filter逻辑完全错误，将外键字段转为字符串后与关联表的name字段比较，不符合原始SQL的关联逻辑

错误分析与修正代码 #

错误点 #

1. join方法使用错误：SQLAlchemy的join需要逐个关联表，每个join操作对应一个关联表和关联条件，不能一次性传入多个外键字段。
2. filter条件逻辑错误：原始SQL是通过外键ID关联对应表的主键，而非将外键ID转为字符串与关联表的name字段匹配，且关联逻辑应在join中处理，无需用filter实现。
3. 查询字段不符：原始SQL仅查询指定字段，你之前的代码查询了整个模型对象，与需求不符。

修正后的代码（手动指定关联条件） #

def get_list(db: Session):
    return db.query(
        model.ListaProduccion.id,
        model.Estado.name,
        model.Orquestador.name,
        model.Comentario.name,
        model.Tipo.name,
        model.Responsable.name,
        model.Compania.name
    ).select_from(model.ListaProduccion) \
     .join(model.Estado, model.ListaProduccion.estado_id == model.Estado.id) \
     .join(model.Orquestador, model.ListaProduccion.orquestador_id == model.Orquestador.id) \
     .join(model.Comentario, model.ListaProduccion.comentario_id == model.Comentario.id) \
     .join(model.Compania, model.ListaProduccion.compania_id == model.Compania.id) \
     .join(model.Tipo, model.ListaProduccion.tipo_id == model.Tipo.id) \
     .join(model.Responsable, model.ListaProduccion.responsable_id == model.Responsable.id) \
     .all()

更简洁的代码（利用模型已定义的外键关系） #

如果你的模型已经通过relationship定义了关联（比如ListaProduccion类中定义了estado = relationship("Estado")等属性），可以直接通过模型关系关联：

def get_list(db: Session):
    return db.query(
        model.ListaProduccion.id,
        model.Estado.name,
        model.Orquestador.name,
        model.Comentario.name,
        model.Tipo.name,
        model.Responsable.name,
        model.Compania.name
    ).join(model.ListaProduccion.estado) \
     .join(model.ListaProduccion.orquestador) \
     .join(model.ListaProduccion.comentario) \
     .join(model.ListaProduccion.compania) \
     .join(model.ListaProduccion.tipo) \
     .join(model.ListaProduccion.responsable) \
     .all()

内容的提问来源于stack exchange，提问作者SANDRA