如何将玩家胜率统计Python代码转为单条SQLAlchemy查询？

我正在为一个游戏项目构建数据仪表盘，玩家可在游戏中担任特定位置，也可在单场游戏中兼任多个位置。仪表盘需获取以下数据：

• Player表中的所有玩家；
• 玩家出场的游戏数量（单场游戏中多位置出场仅算1次，例如Fred在2场游戏中各担任3个位置，统计结果应为2而非6）；
• 玩家出场游戏的胜率（例如John出场20场游戏，其中11场outcome="Win"，胜率应为0.55）。

我原本的Python代码通过循环执行多次数据库查询实现，但速度极慢，能否通过单条SQLAlchemy查询实现相同输出？

原有慢代码 #

from sqlalchemy import func, select

def get_player_winning_percentages():
    """
    Returns a list of all players who have appeared in a game, along with the
    number of games they've appeared in (regardless of how many times they
    appeared in each game) and the winning percentage of games they've appeared in.
    Output looks like:
    [("John", 20, 0.55), ("Alejandro", 15, 0.75), ("Fred", 13, 0.5), ...]
    """
    # Get all players who have appeared in a game
    players_query = (
        Player.query.join(GameToPlayerPosition)
        .group_by(Player.id)
        .order_by(func.count(Player.id).desc())
    )
    # For each player with an appearance, find the number of games they've appeared in,
    # and the number of winning games they've appeared in.
    data = []
    for player in players_query.all():
        games = Game.query.filter(
            Game.players.any(Player.name == player.name)
        ).count()
        wins = Game.query.filter(
            Game._outcome == "Win",
            Game.players.any(Player.name == player.name)
        ).count()
        data.append({
            "name": player.name, "count": games, "wins": wins
        })
    
    # Find players who haven't appeared in a game and add them to the data
    no_appearances = Player.query.filter(
        ~Player.id.in_(select(players_query.subquery().c.id))
    ).all()
    data += [{"name": player.name, "count": 0, "wins": 0} for player in no_appearances]
    return data

数据库表结构 #

# 使用sqlalchemy和Flask_SQLAlchemy
from my_flask_app import db
from sqlalchemy.ext.associationproxy import association_proxy

class Game(db.Model):
    """
    代表一场游戏。每场游戏可以有多个玩家担任不同位置。
    单个玩家可以在一场游戏中多次出现（即关联多个GameToPlayerPosition记录）。
    """
    id = db.Column(db.Integer, primary_key=True)
    # "Win" 或 "Loss"
    outcome = db.Column(db.String(10), nullable=False)
    ... # 其他字段
    positions = db.relationship(
        "GameToPlayerPosition", back_populates="game", cascade="all, delete"
    )
    players = association_proxy(
        "positions",
        "player",
        creator=lambda player, position: GameToPlayerPosition(player=player, position=position),
    )

class GameToPlayerPosition(db.Model):
    """
    连接Game和Player的关联表，附加`position`字段信息。
    """
    game_id = db.Column(db.ForeignKey("game.id"), primary_key=True)
    player_id = db.Column(db.ForeignKey(f"player.id"), primary_key=True)
    position = db.Column(db.String(10), nullable=False)
    ... # 其他字段
    game = db.relationship("Game", back_populates="positions")
    player = db.relationship("Player", back_populates="game_positions")

class Player(db.Model):
    """游戏中的玩家"""
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(60), nullable=False, unique=True)
    ... # 其他字段
    game_positions = db.relationship(
        "GameToPlayerPosition", back_populates="player", cascade="all, delete"
    )
    games = association_proxy("game_positions", "game")

可以通过左连接+分组聚合实现一次查询获取所有数据，彻底解决循环查询的性能问题。核心逻辑如下：

1. 从Player表出发，左连接到中间关联表GameToPlayerPosition，再左连接到Game表；
2. 按玩家ID和姓名分组，用func.count(func.distinct(Game.id))统计玩家参与的唯一游戏数；
3. 用case表达式统计获胜场次，当游戏结果为"Win"时计1，否则计0；
4. 用coalesce和nullif处理0场游戏的情况，避免除以0的错误，此时胜率设为0.0。

优化后的代码 #

from sqlalchemy import func, case

def get_player_winning_percentages():
    # 构建单条查询，一次性获取所有玩家的统计数据
    query = (
        Player.query
        .outerjoin(GameToPlayerPosition, Player.id == GameToPlayerPosition.player_id)
        .outerjoin(Game, GameToPlayerPosition.game_id == Game.id)
        .group_by(Player.id, Player.name)
        .with_entities(
            Player.name,
            func.count(func.distinct(Game.id)).label("game_count"),
            func.sum(case((Game.outcome == "Win", 1), else_=0)).label("win_count"),
            # 计算胜率，处理0场游戏的边界情况
            func.coalesce(
                func.sum(case((Game.outcome == "Win", 1), else_=0)) / func.nullif(func.count(func.distinct(Game.id)), 0),
                0.0
            ).label("win_percentage")
        )
        .order_by(func.count(func.distinct(Game.id)).desc())
    )
    
    # 执行查询并转换为目标输出格式
    results = query.all()
    return [
        (name, game_count, round(win_percentage, 2)) 
        for name, game_count, win_count, win_percentage in results
    ]

关键细节说明 #

• 左连接：确保从未参与过游戏的玩家也被包含在结果中，这类玩家的game_count和win_count会自动为0；
• distinct(Game.id)：强制单场游戏无论玩家担任多少位置，都只计数1次；
• case表达式：精准统计获胜场次，替代原有多次过滤查询的逻辑；
• coalesce + nullif：当玩家没有出场记录时，nullif把0转为NULL，coalesce再将NULL替换为0.0，避免除以0的错误；
• with_entities：明确指定返回字段，减少不必要的数据传输，提升查询效率。

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