如何用SQLAlchemy建立Orders与Jobs表关联并按状态查询数据

一、修正表关联定义 #

你的Jobs类中关联关系的命名有误，OrderName是Orders表的字段，而relationship应关联Orders模型实例，建议修改为规范命名：

修正后的Jobs类 #

from sqlalchemy import Column, String, Integer, ForeignKey
from sqlalchemy.orm import relationship

from MyProject.db.base import Base

class Jobs(Base):
    __tablename__ = 'Jobs'

    JobId = Column(Integer, primary_key=True)
    JobName = Column(String)
    BatchId = Column(Integer)
    OrderId = Column(Integer, ForeignKey('Orders.OrderId'))
    # 关联Orders模型，backref会让Orders自动拥有jobs属性
    order = relationship("Orders", backref='jobs')

    def __init__(self, JobName, BatchId, OrderId):
        self.JobName = JobName
        self.BatchId = BatchId
        self.OrderId = OrderId

修正后的Orders类（可选补充显式关联，不影响功能） #

from sqlalchemy import Column, String, Integer
from sqlalchemy.orm import relationship

from MyProject.db.base import Base

class Orders(Base):
    __tablename__ = 'Orders'

    OrderId = Column(Integer, primary_key=True)
    OrderName = Column(String)
    OrderStatus = Column(Integer)
    # 显式关联Jobs，和backref效果一致，二选一即可
    # jobs = relationship("Jobs", back_populates="order")

    def __init__(self, OrderName, OrderStatus):
        self.OrderName = OrderName
        self.OrderStatus = OrderStatus

二、按OrderStatus筛选Jobs并生成目标字典 #

假设你已通过create_engine和sessionmaker创建了数据库会话Session，以下是完整的查询和格式转换逻辑：

from sqlalchemy.orm import Session
from MyProject.db.models import Jobs, Orders  # 替换为你的模型实际导入路径

def get_jobs_by_order_status(target_status: int) -> dict:
    # 创建数据库会话
    session = Session()
    try:
        # 关联Orders表，筛选指定OrderStatus的Jobs数据
        jobs = session.query(Jobs).join(Orders).filter(Orders.OrderStatus == target_status).all()
        
        # 生成目标格式的API返回字典
        result = {}
        for job in jobs:
            result[job.JobId] = {
                "OrderName": job.order.OrderName,  # 通过关联的order实例获取订单名称
                "JobName": job.JobName,
                "OrderId": job.OrderId,
                "BatchId": job.BatchId
            }
        return result
    finally:
        # 确保会话关闭，避免资源泄漏
        session.close()

# 调用示例：筛选OrderStatus为1的Jobs数据
api_response = get_jobs_by_order_status(1)
print(api_response)

关键说明 #

• 关联逻辑对应：如果你熟悉Perl的DBIx::Class，SQLAlchemy的relationship对应DBIx::Class的has_many/belongs_to，关联查询的逻辑思路一致。
• 关联查询：使用join(Orders)关联两张表，通过filter(Orders.OrderStatus == target_status)实现按订单状态筛选任务。
• 关联字段访问：通过job.order可直接访问当前Job关联的Orders实例，进而获取OrderName字段，这是SQLAlchemy ORM的核心特性。
• 会话管理：使用try...finally确保数据库会话无论是否出错都会关闭，避免资源泄漏。

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