geovisualisierung/projects/project-3/backend/routers/bikepoints.py

62 lines
1.4 KiB
Python
Raw Normal View History

2020-12-20 01:36:36 +01:00
import json
from typing import List
import requests
from fastapi import APIRouter
from pydantic import BaseModel
from api_database import UPSTREAM_BASE_URL
router = APIRouter(prefix="/bikepoints", tags=["bikepoints"])
class BikepointStatus(BaseModel):
NbBikes: int
NbEmptyDocks: int
NbDocks: int
class Bikepoint(BaseModel):
id: str
commonName: str
lat: float
lon: float
status: BikepointStatus
def bikepoint_mapper(bikepoint):
mapped_point = {
"id": bikepoint['id'].removeprefix("BikePoints_"),
"url": bikepoint['url'],
"commonName": bikepoint['commonName'],
"lat": bikepoint['lat'],
"lon": bikepoint['lon']
}
props = list(filter(
lambda p: p['key'] in ["NbBikes", "NbEmptyDocks", "NbDocks"],
bikepoint['additionalProperties']
))
mapped_point['status'] = {prop['key']: int(prop['value']) for prop in props}
return mapped_point
@router.get(
"/",
tags=["upstream"],
response_model=List[Bikepoint]
)
def get_all():
bikepoints = json.loads(requests.get(UPSTREAM_BASE_URL + "/BikePoint").text)
mapped_points = list(map(bikepoint_mapper, bikepoints))
return mapped_points
@router.get(
"/{id}",
tags=["upstream"],
response_model=Bikepoint
)
def get_single(id: int):
bikepoint = json.loads(requests.get(UPSTREAM_BASE_URL + f"/BikePoint/BikePoints_{id}").text)
return bikepoint_mapper(bikepoint)