sqlalchemy-限制加入的结果

Question

模型：

class Team(Base):
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)

    players = relationship("Player", backref="team")

class Player(Base):
    id = Column(Integer, primary_key=True)
    name = Column(String(255), unique=True)
    team_id = Column(Integer, ForeignKey("team.id"))
    positions = relationship("Position", backref="player")


class Position(Base):
    id = Column(Integer(), primary_key=True)
    name = Column(String(255), unique=True)
    player_id = Column(Integer, ForeignKey("player.id"))
    goals = relationship("Goal", backref="position")


class Goal(Base):
    id = Column(Integer(), primary_key=True)
    distance = Column(Integer)
    position_id = Column(Integer, ForeignKey("position.id"))

# Query to get all goals of all players of a team
query = (
    select(Team)
    .select_from(Player, Position, Goal)
    .options(joinedload(Team.players))
    .options(
        joinedload(
            Team.players,
            Player.positions,
        )
    )
    .options(
        joinedload(
            Team.players,
            Player.positions,
            Position.goals,
        )
    )
result = await db.execute(query)
response = result.scalar()

从上面查询的样本JSON输出，

{
    "id": 3,
    "players": [
        {
            "id": 3,
            "positions": []
        },
        {
            "id": 5,
            "positions": [
                {
                    "id": 7,
                    "goals": [
                        {
                            "id": 13,
                        }
                    ]
                }
            ]
        },
        {
            "id": 1,
            "positions": [
                {
                    "id": 1,
                    "goals": [
                        {
                            "id": 16,
                        },
                        {
                            "id": 15,
                        },
                        {
                            "id": 14,
                        }
                    ]
                },
                {
                    "id": 2,
                    "goals": [
                        {
                            "id": 4,
                        }
                    ]
                }
            ]
        }
    ]
}

从示例JSON，我们可以清楚地看到播放器ID = 1的多个目标。

现在，我需要将查询限制为每个玩家的最后一个目标，而不是该玩家的所有目标。

因此，我尝试了

subquery = (
    select(Goal)
    .order_by(Goal.id.desc())
    .limit(1)
    .subquery()
    .lateral()
)

query = (
    select(Team)
    .select_from(Player, Position, Goal)
    .options(joinedload(Team.players))
    .options(
        joinedload(
            Team.players,
            Player.positions,
        )
    )
    .outerjoin(subquery)
    .options(
        contains_eager(
            Team.players,
            Player.positions,
            Position.goals,
            alias=subquery,
        )
    )
result = await db.execute(query)
response = result.scalar()

从上面查询的JSON输出进行采样，

{
    "id": 3,
    "players": [
        {
            "id": 3,
            "positions": []
        },
        {
            "id": 5,
            "positions": [
                {
                    "id": 7,
                    "goals": [
                        {
                            "id": 16,
                        }
                    ]
                }
            ]
        },
        {
            "id": 1,
            "positions": [
                {
                    "id": 1,
                    "goals": [
                        {
                            "id": 16,
                        }
                    ]
                },
                {
                    "id": 2,
                    "goals": [
                         {
                            "id": 16,
                        }
                    ]
                }
            ]
        }
    ]
}

这是任何玩家的最后一个目标，但不是相应球员的最后一个目标。

goal.position_id == position.id之类的过滤器 in outerajoin或subquery 不起作用或导致错误。

编辑：

看来我需要 populate_existing（），但在新选择方法中不可用。

编辑2：

要简化这些查询，我还考虑创建last_goal_id column position表并更新position表以存储最后的ID插入目标。外国钥匙在两张桌子中彼此正常吗？ 目标将具有position_id和位置将具有last_goal_id。

Answer 1

preamble

首先，我相信下面的行不应该是查询的一部分，因为它将创建笛卡尔产品。执行查询时，请查找SQLalchemy警告：

.select_from(Player, Position, Goal)  # DELETE this as it creates cartesian product

其次，您可以在某种程度上简化原始查询。下面的查询与您的问题相当：

# Query to get all goals of all players of a team
query1 = (
    select(Team)
    # .select_from(Player, Position, Goal)  # DELETE this as it creates cartesian product
    .options(
         joinedload(Team.players)
        .joinedload(Player.positions)
        .joinedload(Position.goals)
    )
)

---

contains_eager作为加入load

也可以通过a）加入来实现。明确的相关表明确，b）向SQLalchemy暗示查询已经包含了所需关系：

query2 = (
    select(Team)
    .outerjoin(Team.players)
    .outerjoin(Player.positions)
    .outerjoin(Position.goals)
    .options(contains_eager(
        Team.players, 
        Player.positions, 
        Position.goals,
    ))  
)

---

解决方案：，

鉴于我们现在可以更明确地在关系上加入条件，因此实现查询的一种方法是下图：

# subquery to use in the join for getting only the last 1 goal for each Position
subq = (
    select(Goal.id.label("last_goal_id"))
    .filter(Goal.position_id == Position.id)
    .order_by(Goal.id.desc())
    .limit(1)
    .scalar_subquery()
    .correlate(Position)
)

query3 = (
    select(Team)
    .outerjoin(Team.players)
    .outerjoin(Player.positions)
    .outerjoin(Goal, Goal.id == subq)  # use the JOIN which includes ONLY last Goal, but ...
    .options(contains_eager(
        Team.players, 
        Player.positions, 
        Position.goals,  # ... tell sqlalchemy that we actually loaded ALL `.goals`
    ))
)

生成以下sql（sqlite）：

SELECT goal.id,
       goal.distance,
       goal.position_id,
       position.id AS id_1,
       position.name,
       position.player_id,
       player.id AS id_2,
       player.name AS name_1,
       player.team_id,
       team.id AS id_3,
       team.name AS name_2
FROM team
LEFT OUTER JOIN player ON team.id = player.team_id
LEFT OUTER JOIN position ON player.id = position.player_id
LEFT OUTER JOIN goal ON goal.id =
  (SELECT goal.id AS last_goal_id
   FROM goal
   WHERE goal.position_id = position.id
   ORDER BY goal.id DESC
   LIMIT 1)

---

替代解决方案：

您也可以做什么来创建hybrid_property具有计算的列，哪些指向最后一个goal.id per 位置并使用它来定义一个关系，该关系仅包含列表中的最后一个目标：

class Position(Base):
    __tablename__ = "position"
    id = Column(Integer(), primary_key=True)
    name = Column(String(255), unique=True)
    player_id = Column(Integer, ForeignKey("player.id"))
    goals = relationship("Goal", backref="position")

    @hybrid_property
    def last_goal_id(self):
        ...

    @last_goal_id.expression
    def last_goal_id(cls):
        stmt = (
            select(Goal.id.label("last_goal_id"))
            # .filter(Goal.position_id == Position.id)
            .filter(Goal.position_id == cls.id)
            .order_by(Goal.id.desc())
            .limit(1)
            .scalar_subquery()
            .correlate(cls)
            # .correlate_except(Goal)
        )
        return stmt

    last_goals = relationship(
        lambda: Goal,
        primaryjoin=lambda: and_(
            Goal.position_id == Position.id,
            Goal.id == Position.last_goal_id,
        ),
        viewonly=True,
        uselist=True,
    )

在这种情况下，您可以使用以下查询，但是您不应导航position.goals关系，因为它将加载整个列表。 JSON密钥的名称也将有所不同。

query1 = (
    select(Team)
    .options(
        joinedload(Team.players)
        .joinedload(Player.positions)
        .joinedload(Position.last_goals)  # use `.last_goals` instead of `.goals`
    )
)

注意：我个人最喜欢它，因为它是干净且明确的。

您甚至可以混合技术以获取双方：使用.last_goals关系，但是 trick sa认为这是满载的.goals：

query2 = (
    select(Team)
    .outerjoin(Team.players)
    .outerjoin(Player.positions)
    .outerjoin(Position.last_goals)  # join via `.last_goals` relationship join, but ...
    .options(contains_eager(
        Team.players, 
        Player.positions, 
        Position.goals,  # ... tell sqlalchemy that we actually loaded `.goals`
    ))
)

Answer 2

我认为通过使用 在条款上删除从>目标对象检索到的重复行的

query = (
    # Select from Goal and join all the required tables
    select(Goal)
    .join(Goal.position)
    .join(Position.player)
    .join(Player.team)
    # Remove duplicate rows based on the Player id
    .distinct(Player.id)   
    # Order by `Player.id` (required for distinct) and descending on the goal_id to have the latest added goals (newest) first
    .order_by(Player.id, Goal.id.desc())   
    )

条款

{
    "id": 3,
    "players": [
        {
            "id": 5,
            "positions": [
                {
                    "id": 7,
                    "goals": [
                        {
                            "id": 13,
                        }
                    ]
                }
            ]
        },
        {
            "id": 1,
            "positions": [
                {
                    "id": 1,
                    "goals": [
                        {
                            "id": 16,
                        }
                    ]
                }
            ]
        }
    ]
}

不同 3缺少，因为他没有目标。

您还可以在上的条款上使用diftest of thround of的转动查询：

query = (
    # Select all the required tables
    select(Team, Player, Position, Goal)
    # outerjoin all required tables resulting in a `LEFT OUTER JOIN`
    .outerjoin(Team.players)
    .outerjoin(Player.positions)
    .outerjoin(Position.goals)
    # Remove duplicate rows based on the Player id
    .distinct(Player.id)
    # Order by `Player.id` (required for distinct) and descending on the goal_id to have the latest added goals (newest) first
    .order_by(Player.id, Goal.id.desc())
    )

因为我们现在是从team开始的，然后沿着player降低任何目标还包括以下数据：

{
    "id": 3,
    "players": [
        {
            "id": 3,
            "positions": []
        },
        {
            "id": 5,
            "positions": [
                {
                    "id": 7,
                    "goals": [
                        {
                            "id": 13,
                        }
                    ]
                }
            ]
        },
        {
            "id": 1,
            "positions": [
                {
                    "id": 1,
                    "goals": [
                        {
                            "id": 16,
                        }
                    ]
                }
            ]
        }
    ]
}

示例数据

{
    "id": 3,
    "players": [
        {
            "id": 3,
            "positions": []
        },
        {
            "id": 5,
            "positions": [
                {
                    "id": 7,
                    "goals": [
                        {
                            "id": 13,
                        }
                    ]
                }
            ]
        },
        {
            "id": 1,
            "positions": [
                {
                    "id": 1,
                    "goals": [
                        {
                            "id": 16,
                        },
                        {
                            "id": 15,
                        },
                        {
                            "id": 14,
                        }
                    ]
                },
                {
                    "id": 2,
                    "goals": [
                        {
                            "id": 4,
                        }
                    ]
                }
            ]
        }
    ]
}

Answer 3

考虑使用等级，它可能会做您需要的事情，尽管它需要一些查询/子查询，而不是一个大加入Load。

我将有一个子查询来对目标日期进行排名并按职位或玩家进行分区，并将其滤除到等级1的位置。这将为您提供每个位置的最新目标，您可以为此创建一个命令。通过您的主要查询，您可以使用位置ID使用该dict来找到最新的目标。

类似的事情：

# Rank goals by id and position
subquery = select(
    Goal.id.label('goal_id'),
    Goal.position_id,
    func.rank().over(order_by=Goal.id.desc(), partition_by(Goal.position_id)).label('rank'),
).subquery()

# Create dict of {position_id: latest_goal_id} to use as a lookup
latest_goal_query = (
    select(subquery.c.goal_id, subquery.c.position_id)
    .where(subquery.c.rank == 1)
)
latest_goal_ids = {pos_id: goal_id for goal_id, pos_id in session.execute(latest_goals).fetchall()}

# Get goal objects from the IDs
goal_query = select(Goal).where(Goal.id.in_(latest_goals.values()))
goals = {goal.id: goal for goal in session.execute(goal_query).scalars()}

# Map position ID to the latest goal object
latest_goals = {pos_id: goals[goal_id] for pos_id, goal_id in latest_goal_ids.items()}

# Read the team and position, and you can use the position_id to get the latest goal
query = ...

当一个头btw-我曾经尝试在所有内容上加入，直到sqlalchemy的作者告诉我select> selectInload应在可能的情况下使用，因为它只会提取您需要的数据，而加入可能会有大量的重复数据（例如，如果您的团队有20个球员，每个职位有5个职位，我认为加入这一切都会导致每个团队名称被发送2000次，并且每个播放器名称都发送100次）。

---

编辑：column_property只是作为替代解决方案而想到的。不幸的是，我从来没有能够弄清楚如何映射实际目标模型，因此这不是完美的，但是这是如何将最新目标的ID直接添加到播放器模型中的一个示例。

class Player(Base):
    ...
    latest_goal_id = column_property(
        select(Goal.id)
        .where(Goal.position.has(Position.player_id == id)),
        .order_by(Goal.id.desc()).limit(1)
    )

从查询的角度来看，它只是将其视为另一列，因此您可以选择和过滤。