Eseguire custom query in FastAPI e SQLAlchemy

In questo articolo abbiamo visto come connetterci ad un db MySQL con FastAPI e SQLAlchemy usando i model.

Qui invece vedremo come eseguire una custom query, e inviare l'output in formato JSON.

Prima di tutto facciamo questo:

from fastapi import FastAPI
from database import engine
from sqlalchemy import text

app = FastAPI()

@app.get("/")
async def read_root():
    with engine.connect() as connection:
        rs = connection.execute(text("SELECT * FROM utenti"))
        print(type(rs))
    return {"ciao"}

Questo ci stamperà in console:

<class 'sqlalchemy.engine.cursor.LegacyCursorResult'>

Questo ci serve per capire che dobbiamo trasformare l'output per renderlo compatibile con jsonable_encoder.

Il trucco è questo:

from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse

from database import engine
from sqlalchemy import text

app = FastAPI()

@app.get("/")
async def read_root():
    with engine.connect() as connection:
        rs = connection.execute(text("SELECT * FROM utenti"))
        result = rs.mappings().all()
        json_compatible_item_data = jsonable_encoder(result)
        return JSONResponse(content=json_compatible_item_data)

Enjoy!