62 lines
1.4 KiB
Python
62 lines
1.4 KiB
Python
|
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)
|