如何将MySQL查询语句转换为Flask-SQLAlchemy实现？

需求说明 #

需要实现对应以下MySQL逻辑的Flask-SQLAlchemy代码：从allflash_dev表中获取每个name对应的最新date记录。原MySQL语句如下：

select t.date, t.name from `capacitydata`.`allflash_dev` t
inner join (
select name, max(date) as MaxDate
    from `capacitydata`.`allflash_dev` group by name)
    tm on t.name = tm.name and t.date = tm.MaxDate;

已定义对应的Flask-SQLAlchemy模型：

# Creating Models
class Block(db.Model):
    __tablename__ = "allflash_dev"
 
    index = db.Column(db.Date, nullable=False, unique=True, primary_key=True)
    date = db.Column(db.Date, nullable=False)
    name = db.Column(db.String(45), nullable=False)
    raw = db.Column(db.String(45), nullable=False)
    free = db.Column(db.String(45), nullable=False)
    frep = db.Column(db.String(45), nullable=False)
    util = db.Column(db.String(45), nullable=False)
    utip = db.Column(db.String(45), nullable=False)

def create_db():
    with app.app_context():
        db.create_all()

# Home route
@app.route("/")
def block():
    details = Block.query.order_by(Block.date.desc())
    return render_template("block.html", details=details)

此前尝试的代码未达到预期效果：

details = db.select([Block.name, db.func.max(Block.date)]).group_by(Block.date, Block.name)

正确实现方法 #

方法一：子查询关联（对应原MySQL逻辑） #

先通过子查询获取每个name的最大date，再关联原表筛选出完整的最新记录：

@app.route("/")
def block():
    # 子查询：获取每个name对应的最大date
    subquery = db.session.query(
        Block.name,
        db.func.max(Block.date).label('max_date')
    ).group_by(Block.name).subquery()
    
    # 关联子查询，过滤出每个name的最新记录
    details = db.session.query(Block).join(
        subquery,
        (Block.name == subquery.c.name) & (Block.date == subquery.c.max_date)
    ).all()
    
    return render_template("block.html", details=details)

方法二：窗口函数（简洁高效） #

利用SQL的ROW_NUMBER()窗口函数，按name分组并按date降序排序，直接取每组的第一条记录：

from sqlalchemy import func, over

@app.route("/")
def block():
    # 定义窗口函数：按name分组，date降序生成行号
    row_num = func.row_number().over(
        partition_by=Block.name,
        order_by=Block.date.desc()
    ).label('row_num')
    
    # 子查询添加行号，筛选行号为1的记录
    subquery = db.session.query(Block, row_num).subquery()
    details = db.session.query(subquery).filter(subquery.c.row_num == 1).all()
    
    # 若需直接返回Block对象，可改为：
    # details = db.session.query(Block).from_statement(
    #     db.select(Block).where(row_num == 1)
    # ).all()
    
    return render_template("block.html", details=details)

说明 #

• 方法一完全匹配原MySQL的关联逻辑，兼容性强，适用于所有支持分组查询的数据库
• 方法二使用窗口函数，代码更简洁，执行效率通常更高，适合支持窗口函数的数据库（如MySQL 8.0+、PostgreSQL等）

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