mysql：列表参数必须只包含元组或字典

Question

你好我正在开发一个烧瓶应用程序，我也在编辑谷歌给出的 python 示例，但我的代码中似乎有错误，

sub_topics = conn.execute(text("SELECT DISTINCT sub_topic FROM text_table WHERE topic=%s"), (selected_topic,)).fetchall()

部分代码返回错误

{"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)

{“错误”：“列表参数必须仅包含元组或字典”}

非常感谢，任何帮助将不胜感激