FastAPI+SQLite调用execute_many报错：需传入mapping而非str

你遇到的TypeError: sqlalchemy.sql.elements.TextClause.bindparams() argument after ** must be a mapping, not str错误，核心原因是对database.execute_many()的参数格式理解有误，同时你的循环逻辑也存在数据收集不全的问题。咱们一步步来修复：

错误根源分析 #

• execute_many()要求values参数是包含多个字典的列表，每个字典对应一条待插入的记录（键为数据库字段名，值为对应字段的内容）。
• 你当前的values是一个单字典，每个值都是列表，完全不符合方法的参数要求，因此触发了类型错误。
• 另外你的循环逻辑里，每次迭代都会覆盖movie_title、year等变量，最终只会保留最后一部电影的数据，这也会导致插入结果不符合预期。

修复后的完整代码 #

import uvicorn
from fastapi import FastAPI
from bs4 import BeautifulSoup
import requests
import re
from databases import Database

app = FastAPI()

@app.get("/")
async def fetch_data():
    database = Database('sqlite+aiosqlite:///project.db')
    await database.connect()

    # 先尝试删除表（如果存在），避免重复创建报错
    try:
        query3 = """DROP TABLE imdb"""
        await database.execute(query=query3)
    except:
        pass  # 表不存在时忽略错误

    query = """CREATE TABLE imdb (id INTEGER PRIMARY KEY, movie_title VARCHAR(100), ratings DOUBLE, year INTEGER, cast VARCHAR(255))"""
    await database.execute(query=query)

    url = 'http://www.imdb.com/chart/top'
    response = requests.get(url)
    soup = BeautifulSoup(response.text, "html.parser")

    movies = soup.select('td.titleColumn')
    cast_list = [a.attrs.get('title') for a in soup.select('td.titleColumn a')]
    ratings_list = [float(b.attrs.get('data-value')) for b in soup.select('td.posterColumn span[name=ir]')]

    # 初始化列表存储每部电影的记录
    movie_records = []
    for index in range(0, len(movies)):
        movie_string = movies[index].get_text()
        cleaned_movie = (' '.join(movie_string.split()).replace('.', ''))
        movie_title = cleaned_movie[len(str(index+1))+1:-7]  # 修正索引计算
        year = int(re.search('\((.*?)\)', movie_string).group(1))
        movie_id = index + 1  # 直接用索引+1作为主键，更合理
        current_cast = cast_list[index]
        current_rating = ratings_list[index]

        # 把单条记录整理成字典，添加到列表
        movie_records.append({
            'id': movie_id,
            'movie_title': movie_title,
            'ratings': current_rating,
            'year': year,
            'cast': current_cast
        })

    query2 = "INSERT INTO imdb(id,movie_title,ratings,year, cast) VALUES (:id, :movie_title, :ratings, :year, :cast)"
    # 传入字典列表给execute_many
    await database.execute_many(query=query2, values=movie_records)
    
    query = "SELECT * FROM imdb"
    rows = await database.fetch_all(query=query)
    print('IMDB:', rows)
    await database.disconnect()
    return {"message": "数据爬取并插入成功", "count": len(rows)}

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=5049)

关键修改点说明 #

1. 数据收集逻辑修复：

   • 创建movie_records列表，在循环内将每部电影的信息封装为字典并添加到列表中，确保所有电影数据都被保留。
   • 修正了id的生成逻辑（原来的计算方式会得到空字符串，改用index+1作为自增主键更简单合理）。
   • 将ratings从字符串转为float类型，匹配数据库中DOUBLE字段的类型。
   • 扩大了cast字段的长度（从VARCHAR(100)改为VARCHAR(255)），避免演员信息过长导致插入失败。

2. 适配execute_many参数格式：

   • 将values参数改为movie_records（字典列表），完全符合execute_many的要求，解决了类型错误问题。

3. 额外优化：

   • 给删除表的操作添加了try-except，避免表不存在时抛出错误。
   • 给接口添加了返回值，方便调用时查看结果。

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