python - ** 之后的参数必须是映射，而不是 SQLAlchemy 数据库中的 str

Question

我正在构建一个连接到 SQLite 数据库的 FastAPI。 API 从 IMDB 网站抓取数据，并将其发送到 API 调用的数据库。 下面的代码在第 43 行给出了以下错误:“TypeError: sqlalchemy.sql.elements.TextClause.bindparams() **后面的参数必须是映射，而不是 str”。

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()

    query3 = """DROP TABLE imdb"""
    await database.execute(query=query3)

    query = """CREATE TABLE imdb (id INTEGER PRIMARY KEY, movie_title VARCHAR(100), ratings DOUBLE(1,1), year INTEGER, cast VARCHAR(100))"""
    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 = [a.attrs.get('title') for a in soup.select('td.titleColumn a')]
    ratings = [b.attrs.get('data-value')
    for b in soup.select('td.posterColumn span[name=ir]')]

    for index in range(0, len(movies)):
        
        movie_string = movies[index].get_text()
        movie = (' '.join(movie_string.split()).replace('.', ''))
        movie_title = movie[len(str(index))+1:-7]
        year = re.search('\((.*?)\)', movie_string).group(1)
        id = movie[:len(str(index))-(len(movie))]
      
    values = {'id':[id],'movie_title':[movie_title],'ratings':[ratings],'year':[year],'cast':[cast]}

    query2 = "INSERT INTO imdb(id,movie_title,ratings,year, cast) VALUES (:id, :movie_title, :ratings, :year, :cast)"

    await database.execute_many(query=query2, values=values)
    
    query = "SELECT * FROM imdb"
    rows = await database.fetch_all(query=query)
    print('IMDB:', rows)
    await database.disconnect()

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

我已尝试格式化字典，但仍然遇到相同的错误。我也尝试过 list 和 map() 对象。我应该如何传入映射对象？

Answer 1

查看 Pycharm 中的代码，我在第 42 行看到一些类型不匹配:

await database.execute_many(query=query2, values=values)

特别是 values=values 部分 - 需要一个列表，但是给出一个字典..

查看 execute_many 的实现，我们可以看到它需要一个列表:

async def execute_many(
    self, query: typing.Union[ClauseElement, str], values: list) -> None:
    queries = [self._build_query(query, values_set) for values_set in values]
    async with self._query_lock:
       await self._connection.execute_many(queries)

和build_query:

@staticmethod
def _build_query(
    query: typing.Union[ClauseElement, str], values: typing.Optional[dict]=None) -> ClauseElement:
    if isinstance(query, str):
        query = text(query)

        return query.bindparams(**values) if values is not None else query
    elif values: # <---- trying to unpack
        return query.values(**values) # <---- trying to unpack

    return query

请注意，如果我们只传递一个字典(而不是字典列表)，方法 execute_many 如何生成查询:

def test_execute_many(values):
    queries = [('Some Random Input', values_set) for values_set in values]
    return queries

# Plain old dictionary: 
In [6]: test_execute_many({'a':1, 'b':2})
Out[6]: [('Some Random Input', 'a'), ('Some Random Input', 'b')]

# A list of one dictionary:
In [7]: test_execute_many([{'a':1, 'b':2}])
Out[7]: [('Some Random Input', {'a': 1, 'b': 2})]

现在，_build_query想要使用**values解压给它的值，但是在第一个示例中，它获取一个字符串，并且无法解压它。 .但在第二个示例中它将按预期工作。

我不确定你的其余代码是否有效(因为我没有尝试在本地运行它)，但这个答案涵盖了你所询问的错误😊