如何在Flask-SQLAlchemy中实现Left Join，获取所有作业及当前用户进度

要实现获取所有作业列表，同时关联当前用户的进度数据（无进度的作业也需保留），核心是用左连接（LEFT JOIN）+ 把用户过滤条件放在连接条件中，避免变成内连接或返回无关用户的进度。

原生SQL实现 #

直接通过左连接关联两张表，将用户ID的过滤逻辑写在ON子句里，而非WHERE子句，这样即使当前用户没有该作业的进度记录，作业本身也会被保留，进度字段返回NULL。

示例代码：

SELECT 
    a.*, 
    ua.progress AS user_progress
FROM Assignment a
LEFT JOIN UserAssignments ua 
  ON a.assignmentid = ua.assignmentid 
  AND ua.userid = '当前用户ID'; -- 替换为实际的用户ID变量

ORM实现（主流框架示例） #

Django ORM #

假设你的模型定义如下：

from django.db import models
from django.contrib.auth.models import User

class Assignment(models.Model):
    assignmentid = models.CharField(primary_key=True, max_length=50)
    title = models.CharField(max_length=200)
    # 其他作业相关字段

class UserAssignments(models.Model):
    assignment = models.ForeignKey(Assignment, on_delete=models.CASCADE)
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    progress = models.IntegerField(default=0)

    class Meta:
        unique_together = ('assignment', 'user')  # 确保一个用户对一个作业只有一条进度记录

可以用Subquery来精准获取当前用户的进度，同时保留所有作业：

current_user = request.user  # 获取当前登录用户
assignments = Assignment.objects.annotate(
    user_progress=models.Subquery(
        UserAssignments.objects.filter(
            assignment=models.OuterRef('pk'),
            user=current_user
        ).values('progress')[:1]
    )
).all()

查询结果中，每个Assignment对象会新增user_progress属性：有进度则返回对应值，无进度则为None。

SQLAlchemy #

假设模型定义如下：

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

Base = declarative_base()

class Assignment(Base):
    __tablename__ = 'assignment'
    assignmentid = Column(Integer, primary_key=True)
    title = Column(String(200))
    # 其他字段
    user_assignments = relationship("UserAssignments", back_populates="assignment")

class UserAssignments(Base):
    __tablename__ = 'user_assignments'
    id = Column(Integer, primary_key=True)
    assignmentid = Column(Integer, ForeignKey('assignment.assignmentid'))
    userid = Column(Integer)
    progress = Column(Integer)
    assignment = relationship("Assignment", back_populates="user_assignments")

通过左连接+别名过滤当前用户：

from sqlalchemy import select, aliased
from sqlalchemy.orm import sessionmaker

# 假设已创建session
Session = sessionmaker(bind=engine)
session = Session()

current_user_id = 1  # 替换为当前用户ID
ua_alias = aliased(UserAssignments)

query = select(
    Assignment,
    ua_alias.progress.label('user_progress')
).outerjoin(
    ua_alias,
    (Assignment.assignmentid == ua_alias.assignmentid) & (ua_alias.userid == current_user_id)
)

# 执行查询并处理结果
results = session.execute(query).all()
for assignment, user_progress in results:
    print(f"作业: {assignment.title}, 当前用户进度: {user_progress or '无'}")

为什么之前的查询不符合预期？ #

• 加filter后只返回有进度的作业：因为你用了内连接（或把用户过滤放在WHERE子句），这会自动丢弃没有匹配进度记录的作业。
• 不加filter返回所有用户进度：没有指定用户过滤条件，左连接会关联所有用户的进度记录，导致一个作业对应多条不同用户的进度数据。

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