I am building a FastAPI connected to a SQLite Database. The API scrapes data from the IMDB website and should send it to the database that is called by the API. The below code gives the following error at line 43: "TypeError: sqlalchemy.sql.elements.TextClause.bindparams() argument after ** must be a mapping, not 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)

I have tried formatting the dictionary, but I still get the same error. I have tried list and map() object as well. How should I pass in a mapping object?

CodePudding user response：

Looking at your code in Pycharm, I saw some type mismatch on line 42:

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

Specifically the values=values part - which expects a list, however being given a dictionary..

Looking at the implementation of execute_many we can see it asks for a list:

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)

And 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

Notice how the method execute_many generates the queries if we pass it only a dictionary (and not a list of dictionaries):

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})]

Now, _build_query wants to unpack the values given to it using **values however in the first example, it gets a string, and it cannot unpack it... but in the second example it will work just as expected.

I am not sure if the rest of your code works (as I haven't tried running it locally), but this answer covers the error you're asking about