I am writing an API with FastAPI and am running into this issue with creating a BaseModel for data I am sending

43 Views Asked by At

Basically I tried to use json.dumps() to convert and send a dictionary as such:

{'name': 'Mariner', 'deviceType': 'iPhone 13 Pro', 'latitude': 12.12455, 'longitude': -27.4545622, 'batteryLevel': 28.0, 'positionType': 'Wifi', 'timestamp': 1710790939.373}

That yielded this answer:

({'detail': [{'type': 'model_attributes_type', 'loc': ['body'], 'msg': 'Input should be a valid dictionary or object to extract fields from', 'input': 'grant_type=&username=mariner&password=12345678&scope=&client_id=&client_secret=', 'url': 'https://errors.pydantic.dev/2.5/v/model_attributes_type'}]}, 422

I also tried to run it without converting it to a JSON string, which would output the same thing.

So, on my client's side, I have this so far:

dic = {'name': 'Mariner', 'deviceType': 'iPhone 13 Pro', 'latitude': 12.12455, 'longitude': -27.4545622, 'batteryLevel': 28.0, 'positionType': 'Wifi', 'timestamp': 1710790939.373}

now = connection()

auth = now.authenticate()

message = json.dumps(dic,indent=4)

print(now.publish(auth,message))`

Here is the publish function I made:

def publish(self,auth,json):
        response,status = self.make_request(self.publish_endpoint,method="POST",authheader=auth,params=json)

        return response, status

Here is the make_request:

def make_request(self, url, method='GET', authheader=None, params=None):

        if authheader != None:
            self.headers['Authorization'] = f"Bearer {authheader}"
            
        if method == 'POST':
            try:
                r = requests.post(url, headers=self.headers, data=self.data, timeout=30, json=params)
            except requests.exceptions.Timeout:
                print("Timed out")
                self.make_request(self, url, headers=self.headers, data=self.data, method='GET', authheader=None, params=None)

Then, in terms of the Server side:

This is the model I am using for the API to accept the object.

class DeviceLocation(BaseModel):
    name: str 
    deviceType: str
    latitude: float
    longitude: float
    batteryLevel: float 
    positionType: str
    timestamp: float

And here is how I use the model in the endpoint creation. Currently only trying to print the dictionary out.

@app.post("/logger/publish_data/")
async def save_current_whereabouts(
    current_user: Annotated[User, Depends(get_current_active_user)], item: DeviceLocation
):
    print(item)

    return [{"Status": 200, "owner": current_user.username}]

If there is a better way to get data back and forth, I would be open to suggestions; I've been bashing my head over this for a while now.

0

There are 0 best solutions below