Flask应用SQLAlchemy报错：List argument must consist only of tuples or dictionaries

开发Flask应用时，修改谷歌Python示例代码后，sub_topics相关逻辑返回错误：

{"error":"List argument must consist only of tuples or dictionaries"}

预期获取列表数据，相关代码如下：

import datetime
import logging
import os
from typing import Dict
from sqlalchemy import text
from flask import Flask, render_template, request, Response, jsonify
import sqlalchemy

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

from connect_connector import connect_with_connector
from connect_connector_auto_iam_authn import connect_with_connector_auto_iam_authn
from connect_tcp import connect_tcp_socket
from connect_unix import connect_unix_socket

app = Flask(__name__)

logger = logging.getLogger()


def init_connection_pool() -> sqlalchemy.engine.base.Engine:
    # use a TCP socket when INSTANCE_HOST (e.g. 127.0.0.1) is defined
    if os.environ.get("INSTANCE_HOST"):
        return connect_tcp_socket()

    # use a Unix socket when INSTANCE_UNIX_SOCKET (e.g. /cloudsql/project:region:instance) is defined
    if os.environ.get("INSTANCE_UNIX_SOCKET"):
        return connect_unix_socket()

    # use the connector when INSTANCE_CONNECTION_NAME (e.g. project:region:instance) is defined
    if os.environ.get("INSTANCE_CONNECTION_NAME"):
        # Either a DB_USER or a DB_IAM_USER should be defined. If both are
        # defined, DB_IAM_USER takes precedence.
        return connect_with_connector_auto_iam_authn() if os.environ.get("DB_IAM_USER") else connect_with_connector()

    raise ValueError(
        "Missing database connection type. Please define one of INSTANCE_HOST, INSTANCE_UNIX_SOCKET, or INSTANCE_CONNECTION_NAME"
    )


# create 'votes' table in database if it does not already exist
def migrate_db(db: sqlalchemy.engine.base.Engine) -> None:
    with db.connect() as conn:
        conn.execute(sqlalchemy.text(
            "CREATE TABLE IF NOT EXISTS votes "
            "( vote_id SERIAL NOT NULL, time_cast timestamp NOT NULL, "
            "candidate VARCHAR(6) NOT NULL, PRIMARY KEY (vote_id) );"
        ))
        conn.commit()


# This global variable is declared with a value of `None`, instead of calling
# `init_db()` immediately, to simplify testing. In general, it
# is safe to initialize your database connection pool when your script starts
# -- there is no need to wait for the first request.
db = None


# init_db lazily instantiates a database connection pool. Users of Cloud Run or
# App Engine may wish to skip this lazy instantiation and connect as soon
# as the function is loaded. This is primarily to help testing.
@app.before_first_request
def init_db() -> sqlalchemy.engine.base.Engine:
    global db
    db = init_connection_pool()
    migrate_db(db)

    
@app.route('/', methods=['GET', 'POST'])
def index():
    try:
        # Get all topics
        conn = db.connect()
        topics = conn.execute(text("SELECT DISTINCT topic FROM text_table")).fetchall()
        conn.close()
        topics = [t[0] for t in topics]

        if request.method == 'POST':
            selected_topic = request.form.get('selected_topic')
            if selected_topic:
                # Get all subtopics for the selected topic
                conn = db.connect()
                sub_topics = conn.execute(text("SELECT DISTINCT sub_topic FROM text_table WHERE topic=%s"), (selected_topic,)).fetchall()
                conn.close()
                sub_topics = [s[0] for s in sub_topics]

                selected_sub_topic = request.form.get('selected_sub_topic')
                if selected_sub_topic:
                    # Get the text for the selected subtopic
                    conn = db.connect()
                    result = conn.execute(text("SELECT text FROM text_table WHERE topic=%s AND sub_topic=%s"), (selected_topic, selected_sub_topic)).first()
                    conn.close()
                    result = result[0] if result else ''

                    return render_template('index.html', topics=topics, selected_topic=selected_topic, sub_topics=sub_topics, selected_sub_topic=selected_sub_topic, result=result)

                return render_template('index.html', topics=topics, selected_topic=selected_topic, sub_topics=sub_topics)

        return render_template('index.html', topics=topics)

    except Exception as e:
        # Return an error message if there's an exception
        return jsonify(error=str(e)), 500


if __name__ == "__main__":
    app.run(host="127.0.0.1", port=8080, debug=True)

SQLAlchemy的text()构造SQL语句时，错误使用了%s作为参数占位符。%s是Python字符串格式化语法，而非SQLAlchemy官方推荐的参数绑定方式，这会导致SQLAlchemy无法正确解析参数结构，触发参数格式错误。同时migrate_db函数中的SQL语句存在多余的"语法错误。

1. 将所有SQL语句中的%s替换为命名占位符（如:topic、:sub_topic），并通过字典形式传递参数，确保SQLAlchemy能正确识别参数绑定关系
2. 修正migrate_db函数中的SQL语法错误

import datetime
import logging
import os
from typing import Dict
from sqlalchemy import text
from flask import Flask, render_template, request, Response, jsonify
import sqlalchemy

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

from connect_connector import connect_with_connector
from connect_connector_auto_iam_authn import connect_with_connector_auto_iam_authn
from connect_tcp import connect_tcp_socket
from connect_unix import connect_unix_socket

app = Flask(__name__)

logger = logging.getLogger()


def init_connection_pool() -> sqlalchemy.engine.base.Engine:
    if os.environ.get("INSTANCE_HOST"):
        return connect_tcp_socket()

    if os.environ.get("INSTANCE_UNIX_SOCKET"):
        return connect_unix_socket()

    if os.environ.get("INSTANCE_CONNECTION_NAME"):
        return connect_with_connector_auto_iam_authn() if os.environ.get("DB_IAM_USER") else connect_with_connector()

    raise ValueError(
        "Missing database connection type. Please define one of INSTANCE_HOST, INSTANCE_UNIX_SOCKET, or INSTANCE_CONNECTION_NAME"
    )


def migrate_db(db: sqlalchemy.engine.base.Engine) -> None:
    with db.connect() as conn:
        conn.execute(sqlalchemy.text(
            "CREATE TABLE IF NOT EXISTS votes "
            "( vote_id SERIAL NOT NULL, time_cast timestamp NOT NULL, "
            "candidate VARCHAR(6) NOT NULL, PRIMARY KEY (vote_id) )"
        ))
        conn.commit()


db = None


@app.before_first_request
def init_db() -> sqlalchemy.engine.base.Engine:
    global db
    db = init_connection_pool()
    migrate_db(db)

    
@app.route('/', methods=['GET', 'POST'])
def index():
    try:
        # Get all topics
        conn = db.connect()
        topics = conn.execute(text("SELECT DISTINCT topic FROM text_table")).fetchall()
        conn.close()
        topics = [t[0] for t in topics]

        if request.method == 'POST':
            selected_topic = request.form.get('selected_topic')
            if selected_topic:
                # Get all subtopics for the selected topic
                conn = db.connect()
                sub_topics = conn.execute(
                    text("SELECT DISTINCT sub_topic FROM text_table WHERE topic=:topic"),
                    {"topic": selected_topic}
                ).fetchall()
                conn.close()
                sub_topics = [s[0] for s in sub_topics]

                selected_sub_topic = request.form.get('selected_sub_topic')
                if selected_sub_topic:
                    # Get the text for the selected subtopic
                    conn = db.connect()
                    result = conn.execute(
                        text("SELECT text FROM text_table WHERE topic=:topic AND sub_topic=:sub_topic"),
                        {"topic": selected_topic, "sub_topic": selected_sub_topic}
                    ).first()
                    conn.close()
                    result = result[0] if result else ''

                    return render_template('index.html', topics=topics, selected_topic=selected_topic, sub_topics=sub_topics, selected_sub_topic=selected_sub_topic, result=result)

                return render_template('index.html', topics=topics, selected_topic=selected_topic, sub_topics=sub_topics)

        return render_template('index.html', topics=topics)

    except Exception as e:
        return jsonify(error=str(e)), 500


if __name__ == "__main__":
    app.run(host="127.0.0.1", port=8080, debug=True)

SQLAlchemy官方推荐使用命名占位符进行参数绑定，既避免SQL注入风险，也能保证参数解析的正确性；若坚持使用位置参数，可通过元组列表传递（如conn.execute(text("..."), [("value1",), ("value2",)])），但命名参数的可读性和可维护性更强。

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