Skip to content

Commit

Permalink
fix(refactoring): remove refactoring when calling APIs
Browse files Browse the repository at this point in the history
Fix #41
  • Loading branch information
jonperron committed May 20, 2024
1 parent 026b0de commit 14d24c5
Show file tree
Hide file tree
Showing 13 changed files with 470 additions and 286 deletions.
51 changes: 32 additions & 19 deletions navitia_client/client/apis/arrival_apis.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,14 +30,10 @@ def _get_departures(
pagination = Pagination.from_payload(results.json()["pagination"])
return self._get_departure_objects_from_response(raw_results), pagination

def list_arrivals(
def list_arrivals_by_region_id_and_path(
self,
region_id: Optional[str] = None,
resource_path: Optional[str] = None,
region_lon: Optional[float] = None,
region_lat: Optional[float] = None,
lon: Optional[float] = None,
lat: Optional[float] = None,
region_id: str,
resource_path: str,
from_datetime: datetime = datetime.now(),
duration: int = 86400,
depth: int = 1,
Expand All @@ -46,19 +42,36 @@ def list_arrivals(
disable_geojson: bool = False,
direction_type: str = "all",
) -> Tuple[Sequence[Arrival], Pagination]:
request_url: str | None = None
if region_id and resource_path:
# See https://doc.navitia.io/#route-schedules for URL description
# List of route schedules near the resource
request_url = f"{self.base_navitia_url}/coverage/{region_id}/{resource_path}/terminus_schedules"
elif all([region_lon, region_lat, lon, lat]):
# List of objects near the resource, navitia guesses the region from coordinates
request_url = f"{self.base_navitia_url}/coverage/{region_lon};{region_lat}/coords/{lon};{lat}/terminus_schedules"
request_url = f"{self.base_navitia_url}/coverage/{region_id}/{resource_path}/terminus_schedules"

if not request_url:
raise ValueError(
"Region id and resource path or region coordinates and coordinates must be provided."
)
filters = {
"from_datetime": from_datetime,
"duration": duration,
"depth": depth,
"disable_geojson": disable_geojson,
"forbidden_uris[]": forbidden_uris,
"data_freshness": data_freshness,
"direction_type": direction_type,
}

return self._get_departures(request_url, filters)

def list_arrivals_by_coordinates(
self,
region_lon: float,
region_lat: float,
lon: float,
lat: float,
from_datetime: datetime = datetime.now(),
duration: int = 86400,
depth: int = 1,
forbidden_uris: Optional[Sequence[str]] = None,
data_freshness: str = "realtime",
disable_geojson: bool = False,
direction_type: str = "all",
) -> Tuple[Sequence[Arrival], Pagination]:
# List of objects near the resource, navitia guesses the region from coordinates
request_url = f"{self.base_navitia_url}/coverage/{region_lon};{region_lat}/coords/{lon};{lat}/terminus_schedules"

filters = {
"from_datetime": from_datetime,
Expand Down
50 changes: 31 additions & 19 deletions navitia_client/client/apis/departure_apis.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,14 +30,10 @@ def _get_departures(
pagination = Pagination.from_payload(results.json()["pagination"])
return self._get_departure_objects_from_response(raw_results), pagination

def list_departures(
def list_departures_by_region_id_and_path(
self,
region_id: Optional[str] = None,
resource_path: Optional[str] = None,
region_lon: Optional[float] = None,
region_lat: Optional[float] = None,
lon: Optional[float] = None,
lat: Optional[float] = None,
region_id: str,
resource_path: str,
from_datetime: datetime = datetime.now(),
duration: int = 86400,
depth: int = 1,
Expand All @@ -46,19 +42,35 @@ def list_departures(
disable_geojson: bool = False,
direction_type: str = "all",
) -> Tuple[Sequence[Departure], Pagination]:
request_url: str | None = None
if region_id and resource_path:
# See https://doc.navitia.io/#route-schedules for URL description
# List of route schedules near the resource
request_url = f"{self.base_navitia_url}/coverage/{region_id}/{resource_path}/terminus_schedules"
elif all([region_lon, region_lat, lon, lat]):
# List of objects near the resource, navitia guesses the region from coordinates
request_url = f"{self.base_navitia_url}/coverage/{region_lon};{region_lat}/coords/{lon};{lat}/terminus_schedules"
request_url = f"{self.base_navitia_url}/coverage/{region_id}/{resource_path}/terminus_schedules"

if not request_url:
raise ValueError(
"Region id and resource path or region coordinates and coordinates must be provided."
)
filters = {
"from_datetime": from_datetime,
"duration": duration,
"depth": depth,
"disable_geojson": disable_geojson,
"forbidden_uris[]": forbidden_uris,
"data_freshness": data_freshness,
"direction_type": direction_type,
}

return self._get_departures(request_url, filters)

def list_departures_by_coordinates(
self,
region_lon: float,
region_lat: float,
lon: float,
lat: float,
from_datetime: datetime = datetime.now(),
duration: int = 86400,
depth: int = 1,
forbidden_uris: Optional[Sequence[str]] = None,
data_freshness: str = "realtime",
disable_geojson: bool = False,
direction_type: str = "all",
) -> Tuple[Sequence[Departure], Pagination]:
request_url = f"{self.base_navitia_url}/coverage/{region_lon};{region_lat}/coords/{lon};{lat}/terminus_schedules"

filters = {
"from_datetime": from_datetime,
Expand Down
139 changes: 117 additions & 22 deletions navitia_client/client/apis/places_nearby_apis.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,14 +27,10 @@ def _get_places_nearby(
pagination = Pagination.from_payload(results.json()["pagination"])
return self._get_pt_objects_from_response(raw_results), pagination

def list_objects(
def list_objects_by_region_id_and_path(
self,
region_id: Optional[str] = None,
resource_path: Optional[str] = None,
region_lon: Optional[float] = None,
region_lat: Optional[float] = None,
lon: Optional[float] = None,
lat: Optional[float] = None,
region_id: str,
resource_path: str,
distance: int = 500,
type: Sequence[str] = ["stop_area", "stop_point", "poi"],
admin_uri: Optional[Sequence[str]] = None,
Expand All @@ -46,21 +42,120 @@ def list_objects(
count: int = 25,
add_poi_infos: Sequence[str] = ["bss_stands", "car_park"],
) -> Tuple[Sequence[Place], Pagination]:
request_url: str | None = None
if region_id:
# See https://doc.navitia.io/#places-nearby-api for URL description
if resource_path:
# List of objects near the resource
request_url = f"{self.base_navitia_url}/coverage/{region_id}/{resource_path}/places_nearby"
elif lon and lat:
# List of objects near a coordinate
request_url = f"{self.base_navitia_url}/coverage/{region_id}/{lon};{lat}/places_nearby"
elif region_lon and region_lat and lon and lat:
# List of objects near the resource, navitia guesses the region from coordinates
request_url = f"{self.base_navitia_url}/coverage/{region_lon};{region_lat}/coords/{lon};{lat}/places_nearby"
elif lon and lat and not any([region_id, region_lat, region_lon]):
# List of objects near the resource without any region id (same result as above)
request_url = f"{self.base_navitia_url}/coverage/{lon};{lat}/places_nearby"
request_url = f"{self.base_navitia_url}/coverage/{region_id}/{resource_path}/places_nearby"

filters = {
"start_page": start_page,
"count": count,
"depth": depth,
"type[]": type,
"distance": distance,
"disable_geojson": disable_geojson,
"disable_disruption": disable_disruption,
"add_poi_infos[]": add_poi_infos,
}

if admin_uri:
filters["admin_uris[]"] = admin_uri

if filter:
filters["filter"] = filter

return self._get_places_nearby(request_url, filters)

def list_objects_by_region_id_and_coordinates(
self,
region_id: str,
lon: float,
lat: float,
distance: int = 500,
type: Sequence[str] = ["stop_area", "stop_point", "poi"],
admin_uri: Optional[Sequence[str]] = None,
filter: Optional[str] = None,
disable_geojson: bool = False,
disable_disruption: bool = False,
depth: int = 1,
start_page: int = 0,
count: int = 25,
add_poi_infos: Sequence[str] = ["bss_stands", "car_park"],
) -> Tuple[Sequence[Place], Pagination]:
request_url = (
f"{self.base_navitia_url}/coverage/{region_id}/{lon};{lat}/places_nearby"
)

filters = {
"start_page": start_page,
"count": count,
"depth": depth,
"type[]": type,
"distance": distance,
"disable_geojson": disable_geojson,
"disable_disruption": disable_disruption,
"add_poi_infos[]": add_poi_infos,
}

if admin_uri:
filters["admin_uris[]"] = admin_uri

if filter:
filters["filter"] = filter

return self._get_places_nearby(request_url, filters)

def list_objects_by_coordinates(
self,
region_lon: float,
region_lat: float,
lon: float,
lat: float,
distance: int = 500,
type: Sequence[str] = ["stop_area", "stop_point", "poi"],
admin_uri: Optional[Sequence[str]] = None,
filter: Optional[str] = None,
disable_geojson: bool = False,
disable_disruption: bool = False,
depth: int = 1,
start_page: int = 0,
count: int = 25,
add_poi_infos: Sequence[str] = ["bss_stands", "car_park"],
) -> Tuple[Sequence[Place], Pagination]:
request_url = f"{self.base_navitia_url}/coverage/{region_lon};{region_lat}/coords/{lon};{lat}/places_nearby"

filters = {
"start_page": start_page,
"count": count,
"depth": depth,
"type[]": type,
"distance": distance,
"disable_geojson": disable_geojson,
"disable_disruption": disable_disruption,
"add_poi_infos[]": add_poi_infos,
}

if admin_uri:
filters["admin_uris[]"] = admin_uri

if filter:
filters["filter"] = filter

return self._get_places_nearby(request_url, filters)

def list_objects_by_object_coordinates_only(
self,
lon: float,
lat: float,
distance: int = 500,
type: Sequence[str] = ["stop_area", "stop_point", "poi"],
admin_uri: Optional[Sequence[str]] = None,
filter: Optional[str] = None,
disable_geojson: bool = False,
disable_disruption: bool = False,
depth: int = 1,
start_page: int = 0,
count: int = 25,
add_poi_infos: Sequence[str] = ["bss_stands", "car_park"],
) -> Tuple[Sequence[Place], Pagination]:
request_url = f"{self.base_navitia_url}/coverage/{lon};{lat}/places_nearby"

if not request_url:
raise ValueError(
Expand Down
56 changes: 35 additions & 21 deletions navitia_client/client/apis/route_schedules_apis.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,19 +21,15 @@ def _generate_filter_query(filters: dict[str, Any]) -> str:
filter_query = "&".join([f"{key}={value}" for key, value in filters.items()])
return "?" + filter_query if filter_query else ""

def _get_places_nearby(self, url: str, filters: dict) -> Sequence[RouteSchedule]:
def _get_routes_nearby(self, url: str, filters: dict) -> Sequence[RouteSchedule]:
results = self.get_navitia_api(url + self._generate_filter_query(filters))
raw_results = results.json()["route_schedules"]
return self._get_route_schedule_object_from_response(raw_results)

def list_route_schedules(
def list_route_schedules_by_region_id_and_path(
self,
region_id: Optional[str] = None,
resource_path: Optional[str] = None,
region_lon: Optional[float] = None,
region_lat: Optional[float] = None,
lon: Optional[float] = None,
lat: Optional[float] = None,
region_id: str,
resource_path: str,
from_datetime: datetime = datetime.now(),
duration: int = 86400,
depth: int = 1,
Expand All @@ -43,19 +39,37 @@ def list_route_schedules(
disable_geojson: bool = False,
direction_type: str = "all",
) -> Sequence[RouteSchedule]:
request_url: str | None = None
if region_id and resource_path:
# See https://doc.navitia.io/#route-schedules for URL description
# List of route schedules near the resource
request_url = f"{self.base_navitia_url}/coverage/{region_id}/{resource_path}/route_schedules"
elif all([region_lon, region_lat, lon, lat]):
# List of objects near the resource, navitia guesses the region from coordinates
request_url = f"{self.base_navitia_url}/coverage/{region_lon};{region_lat}/coords/{lon};{lat}/route_schedules"
request_url = f"{self.base_navitia_url}/coverage/{region_id}/{resource_path}/route_schedules"

if not request_url:
raise ValueError(
"Region id and resource path or region coordinates and coordinates must be provided."
)
filters = {
"from_datetime": from_datetime,
"duration": duration,
"depth": depth,
"items_per_schedule": items_per_schedule,
"disable_geojson": disable_geojson,
"forbidden_uris[]": forbidden_uris,
"data_freshness": data_freshness,
"direction_type": direction_type,
}

return self._get_routes_nearby(request_url, filters)

def list_route_schedules_by_coordinates(
self,
region_lon: float,
region_lat: float,
lon: float,
lat: float,
from_datetime: datetime = datetime.now(),
duration: int = 86400,
depth: int = 1,
items_per_schedule: int = 1,
forbidden_uris: Optional[Sequence[str]] = None,
data_freshness: str = "base_schedule",
disable_geojson: bool = False,
direction_type: str = "all",
) -> Sequence[RouteSchedule]:
request_url = f"{self.base_navitia_url}/coverage/{region_lon};{region_lat}/coords/{lon};{lat}/route_schedules"

filters = {
"from_datetime": from_datetime,
Expand All @@ -68,4 +82,4 @@ def list_route_schedules(
"direction_type": direction_type,
}

return self._get_places_nearby(request_url, filters)
return self._get_routes_nearby(request_url, filters)
Loading

0 comments on commit 14d24c5

Please sign in to comment.