CODEWITHSUNDEEP
pythonpython-3.xfastapi
user_12
user_12

Reputation: 2129

How to connect to a sqlite3 db file and fetch contents in fastapi?

I have a sqlite.db file which has 5 columns and 10million rows. I have created a api using fastapi, now in one of the api methods I want to connect to that sqlite.db file and fetch content based on certain conditions (based on the columns present). I mostly will be using SELECT and WHERE.

How can I do it by also taking advantage of async requests. I have came across Tortoise ORM but I am not sure how to properly use it to fetch results.

from fastapi import FastAPI, UploadFile, File, Form
from fastapi.middleware.cors import CORSMiddleware

DATABASE_URL = "sqlite:///test.db"


@app.post("/test")
async def fetch_data(id: int):
    query = "SELECT * FROM tablename WHERE ID={}".format(str(id))

    # how can I fetch such query faster from 10 million records while taking advantage of async func
    return  results

Upvotes: 6

Views: 14303

Answers (2)

theroth
theroth

Reputation: 1

An approach to accessing data from a column in a query result set:

You can read a db result set into a Pandas dataframe. From there, you can use dataframe[“column name”] to access the column data which returns a listlike iterable of that column’s data. You can use the dataframe’s built-in to_dict() method for dictionary data.

Upvotes: 0

Yagiz Degirmenci
Yagiz Degirmenci

Reputation: 20708

You are missing a point here, defining a function with async is not enough. You need to use an asynchronous Database Driver to taking the advantage of using a coroutine.

Encode's Databases library is great for this purpose.

pip install databases

You can also install the required database drivers with:

pip install databases[sqlite]

In your case, this should do good.

from fastapi import FastAPI, UploadFile, File, Form
from fastapi.middleware.cors import CORSMiddleware
from databases import Database

database = Database("sqlite:///test.db")


@app.on_event("startup")
async def database_connect():
    await database.connect()


@app.on_event("shutdown")
async def database_disconnect():
    await database.disconnect()


@app.post("/test")
async def fetch_data(id: int):
    query = "SELECT * FROM tablename WHERE ID={}".format(str(id))
    results = await database.fetch_all(query=query)

    return  results

Upvotes: 15

Related Questions