DynamoDB SQL语句执行报错及重复Python代码优化求助

一、SQL语句ValidationException修复 #

Serverless数据库（如AWS Athena、DynamoDB SQL）对语法规范要求更严格，针对你的查询报错，可按以下方式修正：

• 字段名包含空格和特殊字符（?），必须用双引号或反引号包裹，避免被解析为语法符号
• 确保参数绑定格式符合数据库要求，避免SQL注入风险

修正后的SQL示例：

SELECT COUNT(*) 
FROM game_results 
WHERE player_id = :player_id 
  AND "Team A Result?" < "Team B Result?"

二、Python重复代码精简优化 #

1. 提取公共统计函数 #

把重复的查询、统计逻辑封装为可复用函数，避免代码冗余：

def count_player_team_losses(player_id, db_conn, table="game_results"):
    query = """
        SELECT COUNT(*) 
        FROM %s 
        WHERE player_id = %s 
          AND "Team A Result?" < "Team B Result?"
    """
    cursor = db_conn.cursor()
    cursor.execute(query, (table, player_id))
    count = cursor.fetchone()[0] if cursor.rowcount > 0 else 0
    cursor.close()
    return count

2. 批量处理多玩家统计 #

如果需要同时统计多个玩家，改用批量查询减少数据库交互：

def batch_count_player_losses(player_ids, db_conn, table="game_results"):
    placeholders = ", ".join(["%s"] * len(player_ids))
    query = f"""
        SELECT player_id, COUNT(*) 
        FROM {table} 
        WHERE player_id IN ({placeholders}) 
          AND "Team A Result?" < "Team B Result?"
        GROUP BY player_id
    """
    cursor = db_conn.cursor()
    cursor.execute(query, player_ids)
    results = cursor.fetchall()
    cursor.close()
    return {pid: cnt for pid, cnt in results}

3. ORM简化操作（可选） #

若使用SQLAlchemy等ORM框架，可进一步简化代码，无需手写重复SQL：

from sqlalchemy import func, and_

def count_losses_with_orm(player_id, session, GameResultModel):
    return session.query(func.count())\
                  .filter(and_(
                      GameResultModel.player_id == player_id,
                      GameResultModel.team_a_result < GameResultModel.team_b_result
                  )).scalar()

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