Json - Changing Data type input from MySQL in FastAPI

AmirmohammadAliannezhad
July 27, 2024
190 views
1 vote
2 Answers

enter code hereI have "serialize_response" error for this line :

@app.get("/get-sensors/", response_model=List[Data])

and this :

return {"status": "success", "list": data}

how can I fix this!

I want to get data as Dictionary type

the main code is:

import mysql.connector
from fastapi import FastAPI, HTTPException
from typing import Optional, List, Dict, Union
from pydantic import BaseModel
from mysql.connector import Error
from datetime import datetime, timedelta


def get_db_connection():
    return mysql.connector.connect(
        user='*****',
        password='*****',
        host='*****',
        port=*****,
        database='******'
    )


class Data(BaseModel):
    timestamp: Optional[str] = None
    acceleration_x: Optional[float] = None
    acceleration_y: Optional[float] = None
    acceleration_z: Optional[float] = None
    velocity_x: Optional[float] = None
    velocity_y: Optional[float] = None
    velocity_z: Optional[float] = None
    x: Optional[float] = None
    y: Optional[float] = None
    z: Optional[float] = None
    angular_accelerltion_x: Optional[float] = None
    angular_accelerltion_y: Optional[float] = None
    angular_accelerltion_z: Optional[float] = None
    angular_velocity_x: Optional[float] = None
    angular_velocity_y: Optional[float] = None
    angular_velocity_z: Optional[float] = None
    orientation_x: Optional[float] = None
    orientation_y: Optional[float] = None
    orientation_z: Optional[float] = None
    temperature: Optional[float] = None


fibo_sub = FastAPI()


@fibo_sub.post("/insert_sensors/")
def insert_sensors(data: Data):
    db = get_db_connection()
    mycurser = db.cursor()
    insert_query = (
        "INSERT INTO sensors(timestamp,acceleration_x,acceleration_y,acceleration_z,velocity_x,velocity_y,velocity_z,x,y,z,angular_accelerltion_x,angular_accelerltion_y,angular_accelerltion_z,angular_velocity_x,angular_velocity_y,angular_velocity_z,orientation_x,orientation_y,orientation_z,temperature) "
        " VALUES(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s);"
    )
    vals = (data.timestamp, data.acceleration_x, data.acceleration_y, data.acceleration_z, data.velocity_x, data.velocity_y, data.velocity_z, data.x, data.y, data.z, data.angular_accelerltion_x,
            data.angular_accelerltion_y, data.angular_accelerltion_z, data.angular_velocity_x, data.angular_velocity_y, data.angular_velocity_z, data.orientation_x, data.orientation_y, data.orientation_z, data.temperature)
    mycurser.execute(insert_query, vals)
    db.commit()
    return {"status": "success"}


@fibo_sub.get("/get-sensors/", response_model=List[Data])
def get_sensors(start_date: str):
    db = get_db_connection()
    mycurser = db.cursor(dictionary=True)
    start_time = datetime.strptime(start_date, "%Y-%m-%d %H:%M:%S")
    end_time = start_time - timedelta(seconds=60)
    print(start_time, end_time)

    insert_query = "SELECT * FROM sensors WHERE timestamp BETWEEN %s AND %s"
    mycurser.execute(insert_query, (end_time, start_time))

    data = mycurser.fetchall()
    
    return {"status": "success", "list": data}

ERROR :
INFO:     127.0.0.1:7681 - "GET /get-sensors/?start_date=2004-09-10%2000%3A00%3A31 HTTP/1.1" 500 Internal Server Error
ERROR:    Exception in ASGI application
Traceback (most recent call last):
  File "C:UsersAmir_encoreAppDataLocalProgramsPythonPython312Libsite-packagesuvicornprotocolshttphttptools_impl.py", line 399, in run_asgi
    result = await app(  # type: ignore[func-returns-value]
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:UsersAmir_encoreAppDataLocalProgramsPythonPython312Libsite-packagesuvicornmiddlewareproxy_headers.py", line 70, in __call__
    return await self.app(scope, receive, send)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:UsersAmir_encoreAppDataLocalProgramsPythonPython312Libsite-packagesfastapiapplications.py", line 1054, in __call__
    await super().__call__(scope, receive, send)
  File "C:UsersAmir_encoreAppDataLocalProgramsPythonPython312Libsite-packagesstarletteapplications.py", line 123, in __call__
    await self.middleware_stack(scope, receive, send)
  File "C:UsersAmir_encoreAppDataLocalProgramsPythonPython312Libsite-packagesstarlettemiddlewareerrors.py", line 186, in __call__
    raise exc
  File "C:UsersAmir_encoreAppDataLocalProgramsPythonPython312Libsite-packagesstarlettemiddlewareerrors.py", line 164, in __call__
    await self.app(scope, receive, _send)
  File "C:UsersAmir_encoreAppDataLocalProgramsPythonPython312Libsite-packagesstarlettemiddlewareexceptions.py", line 65, in __call__
    await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
  File "C:UsersAmir_encoreAppDataLocalProgramsPythonPython312Libsite-packagesstarlette_exception_handler.py", line 64, in wrapped_app
    raise exc
  File "C:UsersAmir_encoreAppDataLocalProgramsPythonPython312Libsite-packagesstarlette_exception_handler.py", line 53, in wrapped_app
    await app(scope, receive, sender)
  File "C:UsersAmir_encoreAppDataLocalProgramsPythonPython312Libsite-packagesstarletterouting.py", line 756, in __call__
    await self.middleware_stack(scope, receive, send)
  File "C:UsersAmir_encoreAppDataLocalProgramsPythonPython312Libsite-packagesstarletterouting.py", line 776, in app
    await route.handle(scope, receive, send)
  File "C:UsersAmir_encoreAppDataLocalProgramsPythonPython312Libsite-packagesstarletterouting.py", line 297, in handle
    await self.app(scope, receive, send)
  File "C:UsersAmir_encoreAppDataLocalProgramsPythonPython312Libsite-packagesstarletterouting.py", line 77, in app
    await wrap_app_handling_exceptions(app, request)(scope, receive, send)
  File "C:UsersAmir_encoreAppDataLocalProgramsPythonPython312Libsite-packagesstarlette_exception_handler.py", line 64, in wrapped_app
    raise exc
  File "C:UsersAmir_encoreAppDataLocalProgramsPythonPython312Libsite-packagesstarlette_exception_handler.py", line 53, in wrapped_app
    await app(scope, receive, sender)
  File "C:UsersAmir_encoreAppDataLocalProgramsPythonPython312Libsite-packagesstarletterouting.py", line 72, in app
    response = await func(request)
               ^^^^^^^^^^^^^^^^^^^
  File "C:UsersAmir_encoreAppDataLocalProgramsPythonPython312Libsite-packagesfastapirouting.py", line 296, in app
    content = await serialize_response(
              ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:UsersAmir_encoreAppDataLocalProgramsPythonPython312Libsite-packagesfastapirouting.py", line 155, in serialize_response
    raise ResponseValidationError(
fastapi.exceptions.ResponseValidationError: 1 validation errors:
  {'type': 'list_type', 'loc': ('response',), 'msg': 'Input should be a valid list', 'input': {'status': 'success', 'list': [{'id': 8, 'timestamp': datetime.datetime(2004, 9, 10, 0, 0, 
1), 'acceleration_x': Decimal('12.1000'), 'acceleration_y': Decimal('2.8000'), 'acceleration_z': Decimal('23.2000'), 'velocity_x': Decimal('1.1000'), 'velocity_y': Decimal('12.1000'), 'velocity_z': Decimal('1.1000'), 'x': Decimal('1.10'), 'y': Decimal('1.20'), 'z': Decimal('1.20'), 'angular_accelerltion_x': Decimal('2.20'), 'angular_accelerltion_y': Decimal('1.20'), 'angular_accelerltion_z': Decimal('1.20'), 'angular_velocity_x': Decimal('2.20'), 'angular_velocity_y': Decimal('1.20'), 'angular_velocity_z': Decimal('1.20'), 'orientation_x': Decimal('1.20'), 'orientation_y': Decimal('1.20'), 'orientation_z': Decimal('1.20'), 'temperature': Decimal('1.20')}, {'id': 9, 'timestamp': datetime.datetime(2004, 9, 10, 0, 0, 21), 'acceleration_x': Decimal('12.1000'), 'acceleration_y': Decimal('2.8000'), 'acceleration_z': Decimal('0.2000'), 'velocity_x': Decimal('1.1000'), 'velocity_y': Decimal('12.1000'), 'velocity_z': Decimal('1.1000'), 'x': Decimal('4.10'), 'y': Decimal('1.20'), 'z': Decimal('1.20'), 'angular_accelerltion_x': Decimal('2.20'), 'angular_accelerltion_y': Decimal('1.20'), 'angular_accelerltion_z': Decimal('1.20'), 'angular_velocity_x': Decimal('2.20'), 'angular_velocity_y': Decimal('4.20'), 'angular_velocity_z': Decimal('1.20'), 'orientation_x': Decimal('1.20'), 'orientation_y': Decimal('1.20'), 'orientation_z': Decimal('1.20'), 'temperature': Decimal('1.20')}, {'id': 10, 'timestamp': datetime.datetime(2004, 9, 10, 0, 0, 31), 'acceleration_x': Decimal('12.1000'), 'acceleration_y': Decimal('2.8000'), 'acceleration_z': Decimal('0.2000'), 'velocity_x': Decimal('1.1000'), 'velocity_y': Decimal('12.1000'), 'velocity_z': Decimal('1.1000'), 'x': Decimal('4.10'), 'y': Decimal('1.20'), 'z': Decimal('1.20'), 'angular_accelerltion_x': Decimal('2.20'), 'angular_accelerltion_y': Decimal('1.20'), 'angular_accelerltion_z': Decimal('1.20'), 'angular_velocity_x': Decimal('2.20'), 'angular_velocity_y': Decimal('4.20'), 'angular_velocity_z': Decimal('1.20'), 'orientation_x': Decimal('1.20'), 'orientation_y': Decimal('1.20'), 'orientation_z': Decimal('1.20'), 'temperature': Decimal('1.20')}]}}

Answers

Chosen as BEST ANSWER

finally

@fibo_sub.get("/get_data_in_range/")

def get_data_in_range(end_time: str): try: db = get_db_connection() cursor = db.cursor(dictionary=True)

    end_datetime = datetime.strptime(end_time, '%Y-%m-%d %H:%M:%S')

    
    start_datetime = end_datetime - timedelta(seconds=60)

 
    formatted_start_time = start_datetime.strftime('%Y-%m-%d %H:%M:%S')
    formatted_end_time = end_datetime.strftime('%Y-%m-%d %H:%M:%S')

  
    query = (
        "SELECT * FROM sensors WHERE timestamp >= %s AND timestamp <= %s"
    )
    cursor.execute(query, (formatted_start_time, formatted_end_time))
    rows = cursor.fetchall()
    cursor.close()
    db.close()

    if not rows:
        raise HTTPException(status_code=404, detail="No data found in the specified time range.")

    return rows

except mysql.connector.Error as err:
    raise HTTPException(status_code=500, detail=str(err))
except Exception as e:
    raise HTTPException(status_code=500, detail=str(e))

(Edit)

Your response_model type is list[Data], but you are returning dict[str, Any].

You should create pydantic model like MyResponse that will define the structure of your response dictionary. Then change the response_model to MyResponse:


class MyResponse(BaseModel):
    status: str
    list: list[Data]

@app.get("/get-sensors/", response_model=MyResponse)

Working code example:

import datetime
from decimal import Decimal
from typing import Optional

from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel


app = FastAPI()

class Data(BaseModel):
    timestamp: Optional[str] = None
    acceleration_x: Optional[float] = None
    acceleration_y: Optional[float] = None
    acceleration_z: Optional[float] = None
    velocity_x: Optional[float] = None
    velocity_y: Optional[float] = None
    velocity_z: Optional[float] = None
    x: Optional[float] = None
    y: Optional[float] = None
    z: Optional[float] = None
    angular_accelerltion_x: Optional[float] = None
    angular_accelerltion_y: Optional[float] = None
    angular_accelerltion_z: Optional[float] = None
    angular_velocity_x: Optional[float] = None
    angular_velocity_y: Optional[float] = None
    angular_velocity_z: Optional[float] = None
    orientation_x: Optional[float] = None
    orientation_y: Optional[float] = None
    orientation_z: Optional[float] = None
    temperature: Optional[float] = None

class MyResponse(BaseModel):
    status: str
    list: list[Data]


def get_mock_data():
    return [
        {
            'id': 8,
            'timestamp': datetime.datetime(2004, 9, 10, 0, 0, 1),
            'acceleration_x': Decimal('12.1000'),
            'acceleration_y': Decimal('2.8000'),
            'acceleration_z': Decimal('23.2000'),
            'velocity_x': Decimal('1.1000'),
            'velocity_y': Decimal('12.1000'),
            'velocity_z': Decimal('1.1000'),
            'x': Decimal('1.10'),
            'y': Decimal('1.20'),
            'z': Decimal('1.20'),
            'angular_accelerltion_x': Decimal('2.20'),
            'angular_accelerltion_y': Decimal('1.20'),
            'angular_accelerltion_z': Decimal('1.20'),
            'angular_velocity_x': Decimal('2.20'),
            'angular_velocity_y': Decimal('1.20'),
            'angular_velocity_z': Decimal('1.20'),
            'orientation_x': Decimal('1.20'),
            'orientation_y': Decimal('1.20'),
            'orientation_z': Decimal('1.20'),
            'temperature': Decimal('1.20')
        },
    ]

@app.get("/get-sensors/", response_model=MyResponse)
def get_sensors(start_date: str):

    data = get_mock_data()
    
    return {"status": "success", "list": jsonable_encoder(data)}

Please signup or login to give your own answer.

Click here to cancel reply.

Json – Changing Data type input from MySQL in FastAPI

Answers