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.