From bc95fce148f47069009452e0f216b1b944a6d13f Mon Sep 17 00:00:00 2001 From: Jonathan Perron Date: Tue, 11 Nov 2025 12:13:17 +0100 Subject: [PATCH] feat(api): replace dict by request objects for non-pt APIs Changelog: * Add requests objects for non-pt APIs * Update functions used to call APIs to use requests objects rather than dictionnaries * Update tests --- navitia_client/client/apis/arrival_apis.py | 99 ++------ .../client/apis/contributors_apis.py | 35 +-- navitia_client/client/apis/coverage_apis.py | 49 ++-- navitia_client/client/apis/datasets_apis.py | 35 +-- navitia_client/client/apis/departure_apis.py | 159 +++---------- .../client/apis/equipment_report_apis.py | 65 +----- .../client/apis/freefloatings_nearby_apis.py | 97 ++------ navitia_client/client/apis/isochrone_apis.py | 127 +---------- .../client/apis/line_report_apis.py | 40 +--- navitia_client/client/apis/place_apis.py | 27 +-- .../client/apis/places_nearby_apis.py | 5 - .../apis/public_transport_objects_apis.py | 34 +-- .../client/apis/route_schedules_apis.py | 160 ++++--------- .../client/apis/stop_schedules_apis.py | 160 ++++--------- .../client/apis/terminus_schedules_apis.py | 161 ++++--------- .../client/apis/traffic_report_apis.py | 38 +--- navitia_client/entities/request/__init__.py | 45 ++++ navitia_client/entities/request/arrival.py | 62 +++++ .../entities/request/base_entity_request.py | 52 +++++ .../entities/request/contributor.py | 28 +++ navitia_client/entities/request/coverage.py | 28 +++ navitia_client/entities/request/dataset.py | 28 +++ navitia_client/entities/request/departure.py | 67 ++++++ .../entities/request/equipment_report.py | 46 ++++ .../entities/request/freefloatings_nearby.py | 40 ++++ .../entities/request/inverted_geocoding.py | 29 +++ navitia_client/entities/request/isochrone.py | 75 ++++++ navitia_client/entities/request/journey.py | 215 ++++++++++++++++++ .../entities/request/line_report.py | 56 +++++ navitia_client/entities/request/place.py | 52 +++++ .../entities/request/places_nearby.py | 68 ++++++ .../request/public_transport_object.py | 52 +++++ .../entities/request/route_schedule.py | 67 ++++++ .../entities/request/stop_schedule.py | 67 ++++++ .../entities/request/terminus_schedule.py | 67 ++++++ .../entities/request/traffic_report.py | 56 +++++ tests/client/apis/test_arrival_apis.py | 7 +- tests/client/apis/test_contributors_apis.py | 9 +- tests/client/apis/test_coverage_apis.py | 12 +- tests/client/apis/test_datasets_apis.py | 10 +- tests/client/apis/test_departure_apis.py | 7 +- tests/client/apis/test_isochrones_apis.py | 7 +- tests/client/apis/test_places_apis.py | 4 +- .../test_public_transport_objects_apis.py | 6 +- .../client/apis/test_route_schedules_apis.py | 7 +- tests/client/apis/test_stop_schedules_apis.py | 7 +- .../apis/test_terminus_schedules_apis.py | 7 +- 47 files changed, 1541 insertions(+), 1033 deletions(-) create mode 100644 navitia_client/entities/request/__init__.py create mode 100644 navitia_client/entities/request/arrival.py create mode 100644 navitia_client/entities/request/base_entity_request.py create mode 100644 navitia_client/entities/request/contributor.py create mode 100644 navitia_client/entities/request/coverage.py create mode 100644 navitia_client/entities/request/dataset.py create mode 100644 navitia_client/entities/request/departure.py create mode 100644 navitia_client/entities/request/equipment_report.py create mode 100644 navitia_client/entities/request/freefloatings_nearby.py create mode 100644 navitia_client/entities/request/inverted_geocoding.py create mode 100644 navitia_client/entities/request/isochrone.py create mode 100644 navitia_client/entities/request/journey.py create mode 100644 navitia_client/entities/request/line_report.py create mode 100644 navitia_client/entities/request/place.py create mode 100644 navitia_client/entities/request/places_nearby.py create mode 100644 navitia_client/entities/request/public_transport_object.py create mode 100644 navitia_client/entities/request/route_schedule.py create mode 100644 navitia_client/entities/request/stop_schedule.py create mode 100644 navitia_client/entities/request/terminus_schedule.py create mode 100644 navitia_client/entities/request/traffic_report.py diff --git a/navitia_client/client/apis/arrival_apis.py b/navitia_client/client/apis/arrival_apis.py index 113197b..bcfc70a 100644 --- a/navitia_client/client/apis/arrival_apis.py +++ b/navitia_client/client/apis/arrival_apis.py @@ -1,6 +1,6 @@ -from datetime import datetime -from typing import Any, Optional, Sequence, Tuple +from typing import Any, Sequence, Tuple from navitia_client.client.apis.api_base_client import ApiBaseClient +from navitia_client.entities.request.arrival import ArrivalRequest from navitia_client.entities.response import Pagination from navitia_client.entities.response.arrival import Arrival @@ -77,15 +77,7 @@ def list_arrivals_by_region_id_and_path( self, region_id: str, resource_path: str, - from_datetime: datetime = datetime.now(), - duration: int = 86400, - depth: int = 1, - count: int = 10, - start_page: int = 0, - forbidden_uris: Optional[Sequence[str]] = None, - data_freshness: str = "realtime", - disable_geojson: bool = False, - direction_type: str = "all", + request: ArrivalRequest, ) -> Tuple[Sequence[Arrival], Pagination]: """ Retrieves a list of arrivals for a specific region and resource path. @@ -96,45 +88,18 @@ def list_arrivals_by_region_id_and_path( The identifier of the region to fetch arrivals from. resource_path : str The resource path within the region to fetch arrivals for. - from_datetime : datetime, optional - The starting datetime for fetching arrivals (default is current datetime). - duration : int, optional - The duration in seconds for which to fetch arrivals (default is 86400 seconds). - depth : int, optional - The depth of the search (default is 1). - count : int, optional - Maximum number of results (default is 10). - start_page : int, optional - The page number to start from (default is 0). - forbidden_uris : Optional[Sequence[str]], optional - A list of URIs to exclude from the search (default is None). - data_freshness : str, optional - The freshness of the data to fetch, either "realtime" or "base_schedule" (default is "realtime"). - disable_geojson : bool, optional - Whether to disable geoJSON in the response (default is False). - direction_type : str, optional - The direction type of the arrivals to fetch, e.g., "all", "forward", "backward" (default is "all"). + Returns ------- Tuple[Sequence[Arrival], Pagination] A tuple containing a list of Arrival objects and a Pagination object for managing result pages. """ - request_url = f"{self.base_navitia_url}/coverage/{region_id}/{resource_path}/terminus_schedules" - - filters = { - "from_datetime": from_datetime, - "duration": duration, - "depth": depth, - "count": count, - "start_page": start_page, - "disable_geojson": disable_geojson, - "forbidden_uris[]": forbidden_uris, - "data_freshness": data_freshness, - "direction_type": direction_type, - } - - return self._get_arrivals(request_url, filters) + request_url = ( + f"{self.base_navitia_url}/coverage/{region_id}/{resource_path}/arrivals" + ) + + return self._get_arrivals(request_url, request.to_filters()) def list_arrivals_by_coordinates( self, @@ -142,15 +107,7 @@ def list_arrivals_by_coordinates( region_lat: float, lon: float, lat: float, - from_datetime: datetime = datetime.now(), - duration: int = 86400, - depth: int = 1, - count: int = 10, - start_page: int = 0, - forbidden_uris: Optional[Sequence[str]] = None, - data_freshness: str = "realtime", - disable_geojson: bool = False, - direction_type: str = "all", + request: ArrivalRequest, ) -> Tuple[Sequence[Arrival], Pagination]: """ Retrieves a list of arrivals for specific coordinates. @@ -165,24 +122,6 @@ def list_arrivals_by_coordinates( The longitude of the specific location to fetch arrivals for. lat : float The latitude of the specific location to fetch arrivals for. - from_datetime : datetime, optional - The starting datetime for fetching arrivals (default is current datetime). - duration : int, optional - The duration in seconds for which to fetch arrivals (default is 86400 seconds). - depth : int, optional - The depth of the search (default is 1). - count : int, optional - Maximum number of results (default is 10). - start_page : int, optional - The page number to start from (default is 0). - forbidden_uris : Optional[Sequence[str]], optional - A list of URIs to exclude from the search (default is None). - data_freshness : str, optional - The freshness of the data to fetch, either "realtime" or "base_schedule" (default is "realtime"). - disable_geojson : bool, optional - Whether to disable geoJSON in the response (default is False). - direction_type : str, optional - The direction type of the arrivals to fetch, e.g., "all", "forward", "backward" (default is "all"). Returns ------- @@ -190,18 +129,6 @@ def list_arrivals_by_coordinates( A tuple containing a list of Arrival objects and a Pagination object for managing result pages. """ # 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, - "duration": duration, - "depth": depth, - "count": count, - "start_page": start_page, - "disable_geojson": disable_geojson, - "forbidden_uris[]": forbidden_uris, - "data_freshness": data_freshness, - "direction_type": direction_type, - } - - return self._get_arrivals(request_url, filters) + request_url = f"{self.base_navitia_url}/coverage/{region_lon};{region_lat}/coords/{lon};{lat}/arrivals" + + return self._get_arrivals(request_url, request.to_filters()) diff --git a/navitia_client/client/apis/contributors_apis.py b/navitia_client/client/apis/contributors_apis.py index cb97eed..a04a129 100644 --- a/navitia_client/client/apis/contributors_apis.py +++ b/navitia_client/client/apis/contributors_apis.py @@ -1,6 +1,7 @@ from typing import Any, Sequence, Tuple from navitia_client.client.apis.api_base_client import ApiBaseClient +from navitia_client.entities.request.contributor import ContributorRequest from navitia_client.entities.response.contributor import Contributor from navitia_client.entities.response import Pagination @@ -8,19 +9,9 @@ class ContributorsApiClient(ApiBaseClient): """ A client class to interact with the Navitia API for fetching contributors APIs. + Uses the ContributorRequest class to encapsulate query parameters. See https://doc.navitia.io/#contributors - - Methods - ------- - _get_contributors_from_response(raw_contributors_response: Any) -> Sequence[Contributor] - A static method to transform raw API response data into a list of Contributor objects. - - list_contributors(region_id: str, start_page: int = 0, count: int = 25) -> Tuple[Sequence[Contributor], Pagination] - Retrieves a list of contributors for a specified region from the Navitia API. - - get_contributor_on_dataset(region_id: str, dataset_id: str, start_page: int = 0, count: int = 25) -> Tuple[Sequence[Contributor], Pagination] - Retrieves a list of contributors for a specified dataset in a region from the Navitia API. """ @staticmethod @@ -47,7 +38,7 @@ def _get_contributors_from_response( return contributors def list_contributors( - self, region_id: str, start_page: int = 0, count: int = 25 + self, region_id: str, request: ContributorRequest ) -> Tuple[Sequence[Contributor], Pagination]: """ Retrieves a list of contributors for a specific region. @@ -56,18 +47,17 @@ def list_contributors( ---------- region_id : str The identifier of the region to fetch contributors from. - start_page : int, optional - The starting page for pagination (default is 0). - count : int, optional - The number of contributors to fetch per page (default is 25). + request : ContributorRequest + The request object containing query parameters (count, start_page). Returns ------- Tuple[Sequence[Contributor], Pagination] A tuple containing a list of Contributor objects and a Pagination object for managing result pages. """ + url = f"{self.base_navitia_url}/coverage/{region_id}/contributors" results = self.get_navitia_api( - f"{self.base_navitia_url}/coverage/{region_id}/contributors?start_page={start_page}&count={count}" + url + self._generate_filter_query(request.to_filters()) ) raw_results = results.json()["contributors"] pagination = Pagination.from_payload(results.json()["pagination"]) @@ -76,7 +66,7 @@ def list_contributors( ), pagination def get_contributor_on_dataset( - self, region_id: str, dataset_id: str, start_page: int = 0, count: int = 25 + self, region_id: str, dataset_id: str, request: ContributorRequest ) -> Tuple[Sequence[Contributor], Pagination]: """ Retrieves a list of contributors for a specific dataset in a region. @@ -87,18 +77,17 @@ def get_contributor_on_dataset( The identifier of the region to fetch contributors from. dataset_id : str The identifier of the dataset to fetch contributors for. - start_page : int, optional - The starting page for pagination (default is 0). - count : int, optional - The number of contributors to fetch per page (default is 25). + request : ContributorRequest + The request object containing query parameters (count, start_page). Returns ------- Tuple[Sequence[Contributor], Pagination] A tuple containing a list of Contributor objects and a Pagination object for managing result pages. """ + url = f"{self.base_navitia_url}/coverage/{region_id}/contributors/{dataset_id}" results = self.get_navitia_api( - f"{self.base_navitia_url}/coverage/{region_id}/contributors/{dataset_id}?start_page={start_page}&count={count}" + url + self._generate_filter_query(request.to_filters()) ) raw_results = results.json()["contributors"] pagination = Pagination.from_payload(results.json()["pagination"]) diff --git a/navitia_client/client/apis/coverage_apis.py b/navitia_client/client/apis/coverage_apis.py index 84abd3d..c708805 100644 --- a/navitia_client/client/apis/coverage_apis.py +++ b/navitia_client/client/apis/coverage_apis.py @@ -1,6 +1,7 @@ from typing import Any, Sequence, Tuple from navitia_client.client.apis.api_base_client import ApiBaseClient +from navitia_client.entities.request.coverage import CoverageRequest from navitia_client.entities.response.administrative_region import Region from navitia_client.entities.response import Pagination @@ -8,22 +9,9 @@ class CoverageApiClient(ApiBaseClient): """ A client class to interact with the Navitia API for fetching coverage area information. + Uses the CoverageRequest class to encapsulate query parameters. See https://doc.navitia.io/#coverage - - Methods - ------- - _get_regions_from_response(raw_regions_response: Any) -> Sequence[Region] - A static method to transform raw API response data into a list of Region objects. - - list_covered_areas(start_page: int = 0, count: int = 25) -> Tuple[Sequence[Region], Pagination] - Retrieves a list of covered areas from the Navitia API. - - get_coverage_by_region_id(region_id: str, start_page: int = 0, count: int = 25) -> Tuple[Sequence[Region], Pagination] - Retrieves information about a specific region by its ID. - - get_coverage_by_region_coordinates_and_coordinates(lon: float, lat: float, start_page: int = 0, count: int = 25) -> Tuple[Sequence[Region], Pagination] - Retrieves information about a region based on coordinates. """ @staticmethod @@ -47,25 +35,24 @@ def _get_regions_from_response(raw_regions_response: Any) -> Sequence[Region]: return regions def list_covered_areas( - self, start_page: int = 0, count: int = 25 + self, request: CoverageRequest ) -> Tuple[Sequence[Region], Pagination]: """ Retrieves a list of covered areas from the Navitia API. Parameters ---------- - start_page : int, optional - The starting page for pagination (default is 0). - count : int, optional - The number of regions to fetch per page (default is 25). + request : CoverageRequest + The request object containing query parameters (count, start_page). Returns ------- Tuple[Sequence[Region], Pagination] A tuple containing a list of Region objects and a Pagination object for managing result pages. """ + url = f"{self.base_navitia_url}/coverage" results = self.get_navitia_api( - f"{self.base_navitia_url}/coverage?start_page={start_page}&count={count}" + url + self._generate_filter_query(request.to_filters()) ) result_regions = results.json()["regions"] regions = CoverageApiClient._get_regions_from_response(result_regions) @@ -73,7 +60,7 @@ def list_covered_areas( return regions, pagination def get_coverage_by_region_id( - self, region_id: str, start_page: int = 0, count: int = 25 + self, region_id: str, request: CoverageRequest ) -> Tuple[Sequence[Region], Pagination]: """ Retrieves information about a specific region by its ID. @@ -82,18 +69,17 @@ def get_coverage_by_region_id( ---------- region_id : str The identifier of the region to fetch information about. - start_page : int, optional - The starting page for pagination (default is 0). - count : int, optional - The number of regions to fetch per page (default is 25). + request : CoverageRequest + The request object containing query parameters (count, start_page). Returns ------- Tuple[Sequence[Region], Pagination] A tuple containing a list of Region objects and a Pagination object for managing result pages. """ + url = f"{self.base_navitia_url}/coverage/{region_id}" results = self.get_navitia_api( - f"{self.base_navitia_url}/coverage/{region_id}?start_page={start_page}&count={count}" + url + self._generate_filter_query(request.to_filters()) ) result_regions = results.json()["regions"] regions = CoverageApiClient._get_regions_from_response(result_regions) @@ -101,7 +87,7 @@ def get_coverage_by_region_id( return regions, pagination def get_coverage_by_region_coordinates_and_coordinates( - self, lon: float, lat: float, start_page: int = 0, count: int = 25 + self, lon: float, lat: float, request: CoverageRequest ) -> Tuple[Sequence[Region], Pagination]: """ Retrieves information about a region based on coordinates. @@ -112,18 +98,17 @@ def get_coverage_by_region_coordinates_and_coordinates( The longitude of the location to fetch information about. lat : float The latitude of the location to fetch information about. - start_page : int, optional - The starting page for pagination (default is 0). - count : int, optional - The number of regions to fetch per page (default is 25). + request : CoverageRequest + The request object containing query parameters (count, start_page). Returns ------- Tuple[Sequence[Region], Pagination] A tuple containing a list of Region objects and a Pagination object for managing result pages. """ + url = f"{self.base_navitia_url}/coverage/{lon};{lat}" results = self.get_navitia_api( - f"{self.base_navitia_url}/coverage/{lon};{lat}?start_page={start_page}&count={count}" + url + self._generate_filter_query(request.to_filters()) ) result_regions = results.json()["regions"] regions = CoverageApiClient._get_regions_from_response(result_regions) diff --git a/navitia_client/client/apis/datasets_apis.py b/navitia_client/client/apis/datasets_apis.py index e981a66..f3a69b4 100644 --- a/navitia_client/client/apis/datasets_apis.py +++ b/navitia_client/client/apis/datasets_apis.py @@ -1,6 +1,7 @@ from typing import Any, Sequence, Tuple from navitia_client.client.apis.api_base_client import ApiBaseClient +from navitia_client.entities.request.dataset import DatasetRequest from navitia_client.entities.response.dataset import Dataset from navitia_client.entities.response import Pagination @@ -8,19 +9,9 @@ class DatasetsApiClient(ApiBaseClient): """ A client class to interact with the Navitia API for fetching dataset information. + Uses the DatasetRequest class to encapsulate query parameters. See https://doc.navitia.io/#datasets - - Methods - ------- - _get_datasets_from_response(raw_datasets_response: Any) -> Sequence[Dataset] - A static method to transform raw API response data into a list of Dataset objects. - - list_datasets(region_id: str, start_page: int = 0, count: int = 25) -> Tuple[Sequence[Dataset], Pagination] - Retrieves a list of datasets for a specified region from the Navitia API. - - get_dataset_by_id(region_id: str, dataset_id: str, start_page: int = 0, count: int = 25) -> Tuple[Sequence[Dataset], Pagination] - Retrieves information about a specific dataset by its ID within a region. """ @staticmethod @@ -49,7 +40,7 @@ def _get_datasets_from_response(raw_datasets_response: Any) -> Sequence[Dataset] return datasets def list_datasets( - self, region_id: str, start_page: int = 0, count: int = 25 + self, region_id: str, request: DatasetRequest ) -> Tuple[Sequence[Dataset], Pagination]: """ Retrieves a list of datasets for a specified region from the Navitia API. @@ -58,25 +49,24 @@ def list_datasets( ---------- region_id : str The identifier of the region to fetch datasets from. - start_page : int, optional - The starting page for pagination (default is 0). - count : int, optional - The number of datasets to fetch per page (default is 25). + request : DatasetRequest + The request object containing query parameters (count, start_page). Returns ------- Tuple[Sequence[Dataset], Pagination] A tuple containing a list of Dataset objects and a Pagination object for managing result pages. """ + url = f"{self.base_navitia_url}/coverage/{region_id}/datasets" results = self.get_navitia_api( - f"{self.base_navitia_url}/coverage/{region_id}/datasets?start_page={start_page}&count={count}" + url + self._generate_filter_query(request.to_filters()) ) raw_results = results.json()["datasets"] pagination = Pagination.from_payload(results.json()["pagination"]) return DatasetsApiClient._get_datasets_from_response(raw_results), pagination def get_dataset_by_id( - self, region_id: str, dataset_id: str, start_page: int = 0, count: int = 25 + self, region_id: str, dataset_id: str, request: DatasetRequest ) -> Tuple[Sequence[Dataset], Pagination]: """ Retrieves information about a specific dataset by its ID within a region. @@ -87,18 +77,17 @@ def get_dataset_by_id( The identifier of the region to fetch the dataset from. dataset_id : str The identifier of the dataset to fetch. - start_page : int, optional - The starting page for pagination (default is 0). - count : int, optional - The number of datasets to fetch per page (default is 25). + request : DatasetRequest + The request object containing query parameters (count, start_page). Returns ------- Tuple[Sequence[Dataset], Pagination] A tuple containing a list of Dataset objects and a Pagination object for managing result pages. """ + url = f"{self.base_navitia_url}/coverage/{region_id}/datasets/{dataset_id}" results = self.get_navitia_api( - f"{self.base_navitia_url}/coverage/{region_id}/datasets/{dataset_id}?start_page={start_page}&count={count}" + url + self._generate_filter_query(request.to_filters()) ) raw_results = results.json()["datasets"] pagination = Pagination.from_payload(results.json()["pagination"]) diff --git a/navitia_client/client/apis/departure_apis.py b/navitia_client/client/apis/departure_apis.py index 1d45327..6dae00d 100644 --- a/navitia_client/client/apis/departure_apis.py +++ b/navitia_client/client/apis/departure_apis.py @@ -1,51 +1,22 @@ -from datetime import datetime -from typing import Any, Optional, Sequence, Tuple +from typing import Any, Sequence, Tuple from navitia_client.client.apis.api_base_client import ApiBaseClient +from navitia_client.entities.request.departure import DepartureRequest from navitia_client.entities.response import Pagination from navitia_client.entities.response.departure import Departure class DepartureApiClient(ApiBaseClient): """ - A client class to interact with the Navitia API for fetching departure information. + This module provides a client for interacting with the Navitia API to retrieve departure schedules. - See https://doc.navitia.io/#departures + It includes the `DepartureApiClient` class, which allows users to fetch departure information + based on region ID and resource paths or by specifying coordinates. The client uses the + `DepartureRequest` class to encapsulate query parameters. - Methods + Classes ------- - _get_departure_objects_from_response(response: Any) -> Sequence[Departure] - A static method to transform raw API response data into a list of Departure objects. - - _get_departures(url: str, filters: dict) -> Tuple[Sequence[Departure], Pagination] - Fetches departures from the Navitia API based on the provided URL and filters. - - list_departures_by_region_id_and_path( - region_id: str, - resource_path: str, - 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] - Retrieves a list of departures for a specified region and resource path from the Navitia API. - - list_departures_by_coordinates( - 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] - Retrieves a list of departures for a specified location based on coordinates from the Navitia API. + DepartureApiClient + A client for accessing departure schedules from the Navitia API. """ @staticmethod @@ -98,15 +69,7 @@ def list_departures_by_region_id_and_path( self, region_id: str, resource_path: str, - from_datetime: datetime = datetime.now(), - duration: int = 86400, - depth: int = 1, - count: int = 10, - start_page: int = 0, - forbidden_uris: Optional[Sequence[str]] = None, - data_freshness: str = "realtime", - disable_geojson: bool = False, - direction_type: str = "all", + request: DepartureRequest, ) -> Tuple[Sequence[Departure], Pagination]: """ Retrieves a list of departures for a specified region and resource path from the Navitia API. @@ -117,45 +80,21 @@ def list_departures_by_region_id_and_path( The identifier of the region to fetch departures from. resource_path : str The resource path to fetch departures for. - from_datetime : datetime, optional - The starting datetime for fetching departures (default is current time). - duration : int, optional - The duration for which to fetch departures, in seconds (default is 86400 seconds, i.e., 1 day). - depth : int, optional - The depth of the search (default is 1). - count : int, optional - Maximum number of results (default is 10). - start_page : int, optional - The page number to start from (default is 0). - forbidden_uris : Optional[Sequence[str]], optional - A list of URIs to exclude from the search (default is None). - data_freshness : str, optional - The data freshness parameter (default is "realtime"). - disable_geojson : bool, optional - Whether to disable GeoJSON in the response (default is False). - direction_type : str, optional - The direction type for the departures (default is "all"). + request : DepartureRequest + The request object containing query parameters such as from_datetime, + duration, depth, count, start_page, forbidden_uris, data_freshness, + disable_geojson, and direction_type. Returns ------- Tuple[Sequence[Departure], Pagination] A tuple containing a list of Departure objects and a Pagination object for managing result pages. """ - request_url = f"{self.base_navitia_url}/coverage/{region_id}/{resource_path}/terminus_schedules" - - filters = { - "from_datetime": from_datetime, - "duration": duration, - "depth": depth, - "count": count, - "start_page": start_page, - "disable_geojson": disable_geojson, - "forbidden_uris[]": forbidden_uris, - "data_freshness": data_freshness, - "direction_type": direction_type, - } - - return self._get_departures(request_url, filters) + request_url = ( + f"{self.base_navitia_url}/coverage/{region_id}/{resource_path}/departures" + ) + + return self._get_departures(request_url, request.to_filters()) def list_departures_by_coordinates( self, @@ -163,65 +102,31 @@ def list_departures_by_coordinates( region_lat: float, lon: float, lat: float, - from_datetime: datetime = datetime.now(), - duration: int = 86400, - depth: int = 1, - count: int = 10, - start_page: int = 0, - forbidden_uris: Optional[Sequence[str]] = None, - data_freshness: str = "realtime", - disable_geojson: bool = False, - direction_type: str = "all", + request: DepartureRequest, ) -> Tuple[Sequence[Departure], Pagination]: """ - Retrieves a list of departures for a specified location based on coordinates from the Navitia API. + Retrieves a list of departures for a specified region and coordinates from the Navitia API. Parameters ---------- region_lon : float - The longitude of the region to fetch departures for. + The longitude of the region. region_lat : float - The latitude of the region to fetch departures for. + The latitude of the region. lon : float - The longitude of the specific location to fetch departures for. + The longitude of the specific location. lat : float - The latitude of the specific location to fetch departures for. - from_datetime : datetime, optional - The starting datetime for fetching departures (default is current time). - duration : int, optional - The duration for which to fetch departures, in seconds (default is 86400 seconds, i.e., 1 day). - depth : int, optional - The depth of the search (default is 1). - count : int, optional - Maximum number of results (default is 10). - start_page : int, optional - The page number to start from (default is 0). - forbidden_uris : Optional[Sequence[str]], optional - A list of URIs to exclude from the search (default is None). - data_freshness : str, optional - The data freshness parameter (default is "realtime"). - disable_geojson : bool, optional - Whether to disable GeoJSON in the response (default is False). - direction_type : str, optional - The direction type for the departures (default is "all"). + The latitude of the specific location. + request : DepartureRequest + The request object containing query parameters such as from_datetime, + duration, depth, count, start_page, forbidden_uris, data_freshness, + disable_geojson, and direction_type. Returns ------- Tuple[Sequence[Departure], Pagination] A tuple containing a list of Departure objects and a Pagination object for managing result pages. """ - request_url = f"{self.base_navitia_url}/coverage/{region_lon};{region_lat}/coords/{lon};{lat}/terminus_schedules" - - filters = { - "from_datetime": from_datetime, - "duration": duration, - "depth": depth, - "count": count, - "start_page": start_page, - "disable_geojson": disable_geojson, - "forbidden_uris[]": forbidden_uris, - "data_freshness": data_freshness, - "direction_type": direction_type, - } - - return self._get_departures(request_url, filters) + request_url = f"{self.base_navitia_url}/coverage/{region_lon};{region_lat}/coords/{lon};{lat}/departures" + + return self._get_departures(request_url, request.to_filters()) diff --git a/navitia_client/client/apis/equipment_report_apis.py b/navitia_client/client/apis/equipment_report_apis.py index c4338f3..ce19ddb 100644 --- a/navitia_client/client/apis/equipment_report_apis.py +++ b/navitia_client/client/apis/equipment_report_apis.py @@ -1,5 +1,6 @@ -from typing import Optional, Sequence, Tuple +from typing import Sequence, Tuple from navitia_client.client.apis.api_base_client import ApiBaseClient +from navitia_client.entities.request.equipment_report import EquipmentReportRequest from navitia_client.entities.response.equipment_reports import EquipmentReports from navitia_client.entities.response import Pagination @@ -19,22 +20,14 @@ class EquipmentReportsApiClient(ApiBaseClient): list_equipment_reports( region_id: str, - count: int = 10, - depth: int = 1, - filter: Optional[str] = None, - forbidden_uris: Optional[Sequence[str]] = None, - start_page: int = 0, + request: EquipmentReportRequest = EquipmentReportRequest(), ) -> Tuple[Sequence[EquipmentReports], Pagination]: Retrieves equipment reports for a specified region from the Navitia API. list_equipment_reports_with_resource_path( region_id: str, resource_path: str, - count: int = 10, - depth: int = 1, - filter: Optional[str] = None, - forbidden_uris: Optional[Sequence[str]] = None, - start_page: int = 0, + request: EquipmentReportRequest = EquipmentReportRequest(), ) -> Tuple[Sequence[EquipmentReports], Pagination]: Retrieves equipment reports for a specific resource path in a region from the Navitia API. """ @@ -63,11 +56,7 @@ def _get_equipment_reports( def list_equipment_reports( self, region_id: str, - count: int = 10, - depth: int = 1, - filter: Optional[str] = None, - forbidden_uris: Optional[Sequence[str]] = None, - start_page: int = 0, + request: EquipmentReportRequest = EquipmentReportRequest(), ) -> Tuple[Sequence[EquipmentReports], Pagination]: """ Retrieves equipment reports for a specified region from the Navitia API. @@ -78,11 +67,7 @@ def list_equipment_reports( Parameters: region_id (str): The region ID (coverage identifier). - count (int): Elements per page. Defaults to 10. - depth (int): Json response depth. Defaults to 1. - filter (Optional[str]): A filter to refine your request (e.g., 'line.code=A'). - forbidden_uris (Optional[Sequence[str]]): If you want to avoid lines, modes, networks, etc. - start_page (int): The page number. Defaults to 0. + request (EquipmentReportRequest): The request object containing query parameters. Returns: Tuple[Sequence[EquipmentReports], Pagination]: A tuple containing sequences of EquipmentReports objects and Pagination object. @@ -92,28 +77,13 @@ def list_equipment_reports( Therefore, this service is not available by default. """ request_url = f"{self.base_navitia_url}/coverage/{region_id}/equipment_reports" - - filters = { - "count": count, - "depth": depth, - "start_page": start_page, - "forbidden_uris[]": forbidden_uris, - } - - if filter: - filters["filter"] = filter - - return self._get_equipment_reports(request_url, filters) + return self._get_equipment_reports(request_url, request.to_filters()) def list_equipment_reports_with_resource_path( self, region_id: str, resource_path: str, - count: int = 10, - depth: int = 1, - filter: Optional[str] = None, - forbidden_uris: Optional[Sequence[str]] = None, - start_page: int = 0, + request: EquipmentReportRequest = EquipmentReportRequest(), ) -> Tuple[Sequence[EquipmentReports], Pagination]: """ Retrieves equipment reports for a specific resource path in a region from the Navitia API. @@ -125,11 +95,7 @@ def list_equipment_reports_with_resource_path( Parameters: region_id (str): The region ID (coverage identifier). resource_path (str): The resource path (e.g., 'lines/line:A'). - count (int): Elements per page. Defaults to 10. - depth (int): Json response depth. Defaults to 1. - filter (Optional[str]): A filter to refine your request (e.g., 'line.code=A'). - forbidden_uris (Optional[Sequence[str]]): If you want to avoid lines, modes, networks, etc. - start_page (int): The page number. Defaults to 0. + request (EquipmentReportRequest): The request object containing query parameters. Returns: Tuple[Sequence[EquipmentReports], Pagination]: A tuple containing sequences of EquipmentReports objects and Pagination object. @@ -139,15 +105,4 @@ def list_equipment_reports_with_resource_path( Therefore, this service is not available by default. """ request_url = f"{self.base_navitia_url}/coverage/{region_id}/{resource_path}/equipment_reports" - - filters = { - "count": count, - "depth": depth, - "start_page": start_page, - "forbidden_uris[]": forbidden_uris, - } - - if filter: - filters["filter"] = filter - - return self._get_equipment_reports(request_url, filters) + return self._get_equipment_reports(request_url, request.to_filters()) diff --git a/navitia_client/client/apis/freefloatings_nearby_apis.py b/navitia_client/client/apis/freefloatings_nearby_apis.py index 76baa24..3e24c90 100644 --- a/navitia_client/client/apis/freefloatings_nearby_apis.py +++ b/navitia_client/client/apis/freefloatings_nearby_apis.py @@ -1,6 +1,9 @@ -from typing import Any, Dict, Optional, Sequence, Tuple +from typing import Sequence, Tuple from navitia_client.client.apis.api_base_client import ApiBaseClient +from navitia_client.entities.request.freefloatings_nearby import ( + FreefloatingsNearbyRequest, +) from navitia_client.entities.response.free_floating import FreeFloating from navitia_client.entities.response import Pagination @@ -22,18 +25,14 @@ class FreefloatingsNearbyApiClient(ApiBaseClient): region_id: str, lon: float, lat: float, - distance: int = 500, - type: Optional[Sequence[str]] = None, - count: int = 10, + request: FreefloatingsNearbyRequest = FreefloatingsNearbyRequest(), ) -> Tuple[Sequence[FreeFloating], Pagination]: Retrieves free-floating vehicles near coordinates in a specific region from the Navitia API. list_freefloatings_nearby_with_resource_path( region_id: str, resource_path: str, - distance: int = 500, - type: Optional[Sequence[str]] = None, - count: int = 10, + request: FreefloatingsNearbyRequest = FreefloatingsNearbyRequest(), ) -> Tuple[Sequence[FreeFloating], Pagination]: Retrieves free-floating vehicles near a specific resource path in a region from the Navitia API. @@ -42,18 +41,14 @@ class FreefloatingsNearbyApiClient(ApiBaseClient): region_lat: float, lon: float, lat: float, - distance: int = 500, - type: Optional[Sequence[str]] = None, - count: int = 10, + request: FreefloatingsNearbyRequest = FreefloatingsNearbyRequest(), ) -> Tuple[Sequence[FreeFloating], Pagination]: Retrieves free-floating vehicles near coordinates, navitia guesses the region from coordinates. list_freefloatings_nearby_by_coordinates_only( lon: float, lat: float, - distance: int = 500, - type: Optional[Sequence[str]] = None, - count: int = 10, + request: FreefloatingsNearbyRequest = FreefloatingsNearbyRequest(), ) -> Tuple[Sequence[FreeFloating], Pagination]: Retrieves free-floating vehicles near coordinates without any region id. """ @@ -83,9 +78,7 @@ def list_freefloatings_nearby( region_id: str, lon: float, lat: float, - distance: int = 500, - type: Optional[Sequence[str]] = None, - count: int = 10, + request: FreefloatingsNearbyRequest = FreefloatingsNearbyRequest(), ) -> Tuple[Sequence[FreeFloating], Pagination]: """ Retrieves free-floating vehicles near coordinates in a specific region from the Navitia API. @@ -97,9 +90,7 @@ def list_freefloatings_nearby( region_id (str): The region ID (coverage identifier). lon (float): The longitude coordinate. lat (float): The latitude coordinate. - distance (int): Search radius in meters. Defaults to 500. - type (Optional[Sequence[str]]): The type of shared mobility vehicles to return (e.g., bike, scooter, car). - count (int): Maximum number of results to return. Defaults to 10. + request (FreefloatingsNearbyRequest): The request object containing query parameters. Returns: Tuple[Sequence[FreeFloating], Pagination]: A tuple containing sequences of FreeFloating objects and Pagination object. @@ -109,24 +100,13 @@ def list_freefloatings_nearby( Therefore, this service is not available by default. """ request_url = f"{self.base_navitia_url}/coverage/{region_id}/coords/{lon};{lat}/freefloatings_nearby" - - filters: Dict[str, Any] = { - "distance": distance, - "count": count, - } - - if type: - filters["type[]"] = type - - return self._get_freefloatings_nearby(request_url, filters) + return self._get_freefloatings_nearby(request_url, request.to_filters()) def list_freefloatings_nearby_with_resource_path( self, region_id: str, resource_path: str, - distance: int = 500, - type: Optional[Sequence[str]] = None, - count: int = 10, + request: FreefloatingsNearbyRequest = FreefloatingsNearbyRequest(), ) -> Tuple[Sequence[FreeFloating], Pagination]: """ Retrieves free-floating vehicles near a specific resource path in a region from the Navitia API. @@ -137,9 +117,7 @@ def list_freefloatings_nearby_with_resource_path( Parameters: region_id (str): The region ID (coverage identifier). resource_path (str): The resource path (e.g., 'stop_areas/stop_area:XXX'). - distance (int): Search radius in meters. Defaults to 500. - type (Optional[Sequence[str]]): The type of shared mobility vehicles to return (e.g., bike, scooter, car). - count (int): Maximum number of results to return. Defaults to 10. + request (FreefloatingsNearbyRequest): The request object containing query parameters. Returns: Tuple[Sequence[FreeFloating], Pagination]: A tuple containing sequences of FreeFloating objects and Pagination object. @@ -149,16 +127,7 @@ def list_freefloatings_nearby_with_resource_path( Therefore, this service is not available by default. """ request_url = f"{self.base_navitia_url}/coverage/{region_id}/{resource_path}/freefloatings_nearby" - - filters: Dict[str, Any] = { - "distance": distance, - "count": count, - } - - if type: - filters["type[]"] = type - - return self._get_freefloatings_nearby(request_url, filters) + return self._get_freefloatings_nearby(request_url, request.to_filters()) def list_freefloatings_nearby_by_coordinates( self, @@ -166,9 +135,7 @@ def list_freefloatings_nearby_by_coordinates( region_lat: float, lon: float, lat: float, - distance: int = 500, - type: Optional[Sequence[str]] = None, - count: int = 10, + request: FreefloatingsNearbyRequest = FreefloatingsNearbyRequest(), ) -> Tuple[Sequence[FreeFloating], Pagination]: """ Retrieves free-floating vehicles near coordinates, navitia guesses the region from coordinates. @@ -182,9 +149,7 @@ def list_freefloatings_nearby_by_coordinates( region_lat (float): The latitude coordinate for region identification. lon (float): The longitude coordinate for the search center. lat (float): The latitude coordinate for the search center. - distance (int): Search radius in meters. Defaults to 500. - type (Optional[Sequence[str]]): The type of shared mobility vehicles to return (e.g., bike, scooter, car). - count (int): Maximum number of results to return. Defaults to 10. + request (FreefloatingsNearbyRequest): The request object containing query parameters. Returns: Tuple[Sequence[FreeFloating], Pagination]: A tuple containing sequences of FreeFloating objects and Pagination object. @@ -194,24 +159,13 @@ def list_freefloatings_nearby_by_coordinates( Therefore, this service is not available by default. """ request_url = f"{self.base_navitia_url}/coverage/{region_lon};{region_lat}/coords/{lon};{lat}/freefloatings_nearby" - - filters: Dict[str, Any] = { - "distance": distance, - "count": count, - } - - if type: - filters["type[]"] = type - - return self._get_freefloatings_nearby(request_url, filters) + return self._get_freefloatings_nearby(request_url, request.to_filters()) def list_freefloatings_nearby_by_coordinates_only( self, lon: float, lat: float, - distance: int = 500, - type: Optional[Sequence[str]] = None, - count: int = 10, + request: FreefloatingsNearbyRequest = FreefloatingsNearbyRequest(), ) -> Tuple[Sequence[FreeFloating], Pagination]: """ Retrieves free-floating vehicles near coordinates without any region id. @@ -223,9 +177,7 @@ def list_freefloatings_nearby_by_coordinates_only( Parameters: lon (float): The longitude coordinate. lat (float): The latitude coordinate. - distance (int): Search radius in meters. Defaults to 500. - type (Optional[Sequence[str]]): The type of shared mobility vehicles to return (e.g., bike, scooter, car). - count (int): Maximum number of results to return. Defaults to 10. + request (FreefloatingsNearbyRequest): The request object containing query parameters. Returns: Tuple[Sequence[FreeFloating], Pagination]: A tuple containing sequences of FreeFloating objects and Pagination object. @@ -235,13 +187,4 @@ def list_freefloatings_nearby_by_coordinates_only( Therefore, this service is not available by default. """ request_url = f"{self.base_navitia_url}/coord/{lon};{lat}/freefloatings_nearby" - - filters: Dict[str, Any] = { - "distance": distance, - "count": count, - } - - if type: - filters["type[]"] = type - - return self._get_freefloatings_nearby(request_url, filters) + return self._get_freefloatings_nearby(request_url, request.to_filters()) diff --git a/navitia_client/client/apis/isochrone_apis.py b/navitia_client/client/apis/isochrone_apis.py index 284dc63..474591a 100644 --- a/navitia_client/client/apis/isochrone_apis.py +++ b/navitia_client/client/apis/isochrone_apis.py @@ -1,6 +1,6 @@ -from datetime import datetime -from typing import Any, Optional, Sequence +from typing import Sequence from navitia_client.client.apis.api_base_client import ApiBaseClient +from navitia_client.entities.request.isochrone import IsochroneRequest from navitia_client.entities.response.isochrones import Isochrone @@ -16,27 +16,13 @@ class IsochronesApiClient(ApiBaseClient): Internal method to fetch isochrone data based on the provided URL and filters. list_isochrones_with_region_id( - from_: str, region_id: str, - start_datetime: datetime = datetime.now(), - boundary_duration: Sequence[int] = [], - to: Optional[str] = None, - first_section_mode: Optional[Sequence[str]] = None, - last_section_mode: Optional[Sequence[str]] = None, - min_duration: Optional[int] = None, - max_duration: Optional[int] = None + request: IsochroneRequest ) -> Sequence[Isochrone] Fetches isochrones data for a specific region based on various parameters. list_isochrones( - from_: str, - start_datetime: datetime = datetime.now(), - boundary_duration: Sequence[int] = [], - to: Optional[str] = None, - first_section_mode: Optional[Sequence[str]] = None, - last_section_mode: Optional[Sequence[str]] = None, - min_duration: Optional[int] = None, - max_duration: Optional[int] = None + request: IsochroneRequest ) -> Sequence[Isochrone] Fetches isochrones data based on various parameters. """ @@ -65,39 +51,18 @@ def _get_traffic_reports(self, url: str, filters: dict) -> Sequence[Isochrone]: def list_isochrones_with_region_id( self, - from_: str, region_id: str, - start_datetime: datetime = datetime.now(), - boundary_duration: Sequence[int] = [], - to: Optional[str] = None, - first_section_mode: Optional[Sequence[str]] = None, - last_section_mode: Optional[Sequence[str]] = None, - min_duration: Optional[int] = None, - max_duration: Optional[int] = None, + request: IsochroneRequest, ) -> Sequence[Isochrone]: """ Fetches isochrones data for a specific region based on various parameters. Parameters ---------- - from_ : str - The starting point for the isochrone calculation. region_id : str The identifier of the region. - start_datetime : datetime, optional - The starting date and time for the isochrone calculation (default is datetime.now()). - boundary_duration : Sequence[int], optional - List of durations in seconds defining the isochrones boundaries. - to : Optional[str], optional - The ending point for the isochrone calculation. - first_section_mode : Optional[Sequence[str]], optional - Modes of transportation for the first section of the journey. - last_section_mode : Optional[Sequence[str]], optional - Modes of transportation for the last section of the journey. - min_duration : Optional[int], optional - The minimum duration for the isochrone calculation. - max_duration : Optional[int], optional - The maximum duration for the isochrone calculation. + request : IsochroneRequest + The request object containing query parameters. Returns ------- @@ -105,64 +70,19 @@ def list_isochrones_with_region_id( A list of Isochrone objects representing the isochrone data. """ request_url = f"{self.base_navitia_url}/coverage/{region_id}/isochrones" - - filters: dict[str, Any] = { - "datetime": start_datetime.isoformat(), - "boundary_duration[]": boundary_duration, - "from": from_, - } - - if to: - filters["to"] = to - - if min_duration: - filters["min_duration"] = min_duration - - if first_section_mode: - filters["first_section_mode[]"] = first_section_mode - - if last_section_mode: - filters["last_section_mode[]"] = last_section_mode - - if max_duration: - filters["max_duration"] = max_duration - if len(boundary_duration) == 0: - filters.pop("min_duration") - - return self._get_traffic_reports(request_url, filters) + return self._get_traffic_reports(request_url, request.to_filters()) def list_isochrones( self, - from_: str, - start_datetime: datetime = datetime.now(), - boundary_duration: Sequence[int] = [], - to: Optional[str] = None, - first_section_mode: Optional[Sequence[str]] = None, - last_section_mode: Optional[Sequence[str]] = None, - min_duration: Optional[int] = None, - max_duration: Optional[int] = None, + request: IsochroneRequest, ) -> Sequence[Isochrone]: """ Fetches isochrones data based on various parameters. Parameters ---------- - from_ : str - The starting point for the isochrone calculation. - start_datetime : datetime, optional - The starting date and time for the isochrone calculation (default is datetime.now()). - boundary_duration : Sequence[int], optional - List of durations in seconds defining the isochrones boundaries. - to : Optional[str], optional - The ending point for the isochrone calculation. - first_section_mode : Optional[Sequence[str]], optional - Modes of transportation for the first section of the journey. - last_section_mode : Optional[Sequence[str]], optional - Modes of transportation for the last section of the journey. - min_duration : Optional[int], optional - The minimum duration for the isochrone calculation. - max_duration : Optional[int], optional - The maximum duration for the isochrone calculation. + request : IsochroneRequest + The request object containing query parameters. Returns ------- @@ -170,27 +90,4 @@ def list_isochrones( A list of Isochrone objects representing the isochrone data. """ request_url = f"{self.base_navitia_url}/isochrones" - - filters: dict[str, Any] = { - "datetime": start_datetime.isoformat(), - "boundary_duration[]": boundary_duration, - "from": from_, - } - if to: - filters["to"] = to - - if first_section_mode: - filters["first_section_mode[]"] = first_section_mode - - if last_section_mode: - filters["last_section_mode[]"] = last_section_mode - - if min_duration: - filters["min_duration"] = min_duration - - if max_duration: - filters["max_duration"] = max_duration - if len(boundary_duration) == 0: - filters.pop("min_duration") - - return self._get_traffic_reports(request_url, filters) + return self._get_traffic_reports(request_url, request.to_filters()) diff --git a/navitia_client/client/apis/line_report_apis.py b/navitia_client/client/apis/line_report_apis.py index db3377a..d4328ff 100644 --- a/navitia_client/client/apis/line_report_apis.py +++ b/navitia_client/client/apis/line_report_apis.py @@ -1,6 +1,6 @@ -from datetime import datetime from typing import Optional, Sequence, Tuple from navitia_client.client.apis.api_base_client import ApiBaseClient +from navitia_client.entities.request.line_report import LineReportRequest from navitia_client.entities.response.disruption import Disruption from navitia_client.entities.response.line_report import LineReport @@ -16,7 +16,7 @@ class LineReportsApiClient(ApiBaseClient): _get_line_reports(url: str, filters: dict) -> Tuple[Sequence[Disruption], Sequence[LineReport]]: Retrieves line reports from the specified URL with the provided filters. - list_line_reports(region_id: Optional[str] = None, resource_path: Optional[str] = None, since: Optional[datetime] = None, until: Optional[datetime] = None, count: int = 25, depth: int = 1, forbidden_uris: Optional[Sequence[str]] = None, disable_geojson: bool = False) -> Tuple[Sequence[Disruption], Sequence[LineReport]]: + list_line_reports(region_id: Optional[str] = None, resource_path: Optional[str] = None, request: LineReportRequest = LineReportRequest()) -> Tuple[Sequence[Disruption], Sequence[LineReport]]: Lists line reports based on specified criteria. """ @@ -51,12 +51,7 @@ def list_line_reports( self, region_id: Optional[str] = None, resource_path: Optional[str] = None, - since: Optional[datetime] = None, - until: Optional[datetime] = None, - count: int = 25, - depth: int = 1, - forbidden_uris: Optional[Sequence[str]] = None, - disable_geojson: bool = False, + request: LineReportRequest = LineReportRequest(), ) -> Tuple[Sequence[Disruption], Sequence[LineReport]]: """ Lists line reports based on specified criteria. @@ -67,18 +62,8 @@ def list_line_reports( The ID of the region for which to fetch line reports, by default None. resource_path : Optional[str], optional The resource path for line reports, by default None. - since : Optional[datetime], optional - Filter line reports since this date and time, by default None. - until : Optional[datetime], optional - Filter line reports until this date and time, by default None. - count : int, optional - The number of line reports to retrieve, by default 25. - depth : int, optional - The depth of the query, by default 1. - forbidden_uris : Optional[Sequence[str]], optional - List of URIs to forbid, by default None. - disable_geojson : bool, optional - Whether to disable GeoJSON output, by default False. + request : LineReportRequest, optional + The request object containing query parameters. Returns ------- @@ -90,17 +75,4 @@ def list_line_reports( else: request_url = f"{self.base_navitia_url}/coverage/{region_id}/line_reports" - filters = { - "count": count, - "depth": depth, - "disable_geojson": disable_geojson, - "forbidden_uris[]": forbidden_uris, - } - - if since: - filters["since"] = since.isoformat() - - if until: - filters["until"] = until.isoformat() - - return self._get_line_reports(request_url, filters) + return self._get_line_reports(request_url, request.to_filters()) diff --git a/navitia_client/client/apis/place_apis.py b/navitia_client/client/apis/place_apis.py index d8833bb..fe8d186 100644 --- a/navitia_client/client/apis/place_apis.py +++ b/navitia_client/client/apis/place_apis.py @@ -1,5 +1,6 @@ -from typing import Any, Optional, Sequence, Tuple +from typing import Any, Sequence from navitia_client.client.apis.api_base_client import ApiBaseClient +from navitia_client.entities.request.place import PlaceRequest from navitia_client.entities.response.place import Place @@ -42,33 +43,21 @@ def _get_pt_objects_from_response(response: Any) -> Sequence[Place]: def list_places( self, region_id: str, - query: str, - type: Sequence[str] = ["stop_area", "address", "poi", "administrative_region"], - disable_geojson: bool = False, - depth: int = 1, - from_lon_lat: Optional[Tuple[float, float]] = None, + request: PlaceRequest, ) -> Sequence[Place]: """ Retrieves a list of places based on the provided query and region ID from the Navitia API. Parameters: region_id (str): The region ID. - query (str): The query string to search for places. - type (Sequence[str], optional): The types of places to include in the search. - Defaults to ["stop_area", "address", "poi", "administrative_region"]. - disable_geojson (bool, optional): Whether to disable GeoJSON format in the response. - Defaults to False. - depth (int, optional): The depth of data to retrieve. Defaults to 1. - from_lon_lat (Optional[Tuple[float, float]], optional): The longitude and latitude - from which to search for places. Defaults to None. + request (PlaceRequest): The request object containing query parameters. Returns: Sequence[Place]: A list of Place objects matching the query. """ - request_url = f"{self.base_navitia_url}/coverage/{region_id}/places?q={query}&type={type}&disable_geojson={disable_geojson}&depth={depth}" - if from_lon_lat: - request_url += f"&filter={from_lon_lat[0]};{from_lon_lat[1]}" - - results = self.get_navitia_api(request_url) + request_url = f"{self.base_navitia_url}/coverage/{region_id}/places" + results = self.get_navitia_api( + request_url + self._generate_filter_query(request.to_filters()) + ) raw_results = results.json()["places"] return self._get_pt_objects_from_response(raw_results) diff --git a/navitia_client/client/apis/places_nearby_apis.py b/navitia_client/client/apis/places_nearby_apis.py index 2392368..4a43402 100644 --- a/navitia_client/client/apis/places_nearby_apis.py +++ b/navitia_client/client/apis/places_nearby_apis.py @@ -329,11 +329,6 @@ def list_objects_by_object_coordinates_only( """ request_url = f"{self.base_navitia_url}/coverage/{lon};{lat}/places_nearby" - if not request_url: - raise ValueError( - "Region id, region coordinates or coordinates must be provided." - ) - filters = { "start_page": start_page, "count": count, diff --git a/navitia_client/client/apis/public_transport_objects_apis.py b/navitia_client/client/apis/public_transport_objects_apis.py index cebf27f..b1a3e32 100644 --- a/navitia_client/client/apis/public_transport_objects_apis.py +++ b/navitia_client/client/apis/public_transport_objects_apis.py @@ -1,5 +1,8 @@ -from typing import Any, Optional, Sequence +from typing import Any, Sequence from navitia_client.client.apis.api_base_client import ApiBaseClient +from navitia_client.entities.request.public_transport_object import ( + PublicTransportObjectRequest, +) from navitia_client.entities.response.pt_object import PtObject @@ -51,38 +54,21 @@ def _get_pt_objects_from_response(response: Any) -> Sequence[PtObject]: def list_public_transport_objects( self, region_id: str, - query: str, - type: Sequence[str] = [ - "network", - "commercial_mode", - "line", - "route", - "stop_area", - ], - disable_disruption: bool = False, - depth: int = 1, - post_query_filter: Optional[str] = None, + request: PublicTransportObjectRequest, ) -> Sequence[PtObject]: """ Retrieves a list of public transport objects for a specified region from the Navitia API. Parameters: region_id (str): The region ID. - query (str): The search query. - type (Sequence[str], optional): The types of public transport objects to include - in the search. Defaults to ["network", "commercial_mode", "line", "route", "stop_area"]. - disable_disruption (bool, optional): Whether to disable disruption information in the response. - Defaults to False. - depth (int, optional): The depth of data to retrieve. Defaults to 1. - post_query_filter (Optional[str], optional): Additional filtering criteria. Defaults to None. + request (PublicTransportObjectRequest): The request object containing query parameters. Returns: Sequence[PtObject]: A sequence of PtObject objects. """ - request_url = f"{self.base_navitia_url}/coverage/{region_id}/pt_objects?q={query}&type={type}&disable_disruption={disable_disruption}&depth={depth}" - if post_query_filter: - request_url += f"&filter={post_query_filter}" - - results = self.get_navitia_api(request_url) + request_url = f"{self.base_navitia_url}/coverage/{region_id}/pt_objects" + results = self.get_navitia_api( + request_url + self._generate_filter_query(request.to_filters()) + ) raw_results = results.json()["pt_objects"] return self._get_pt_objects_from_response(raw_results) diff --git a/navitia_client/client/apis/route_schedules_apis.py b/navitia_client/client/apis/route_schedules_apis.py index 9062b31..77de3cf 100644 --- a/navitia_client/client/apis/route_schedules_apis.py +++ b/navitia_client/client/apis/route_schedules_apis.py @@ -1,49 +1,15 @@ -from datetime import datetime -from typing import Any, Optional, Sequence +from typing import Any, Sequence from navitia_client.client.apis.api_base_client import ApiBaseClient +from navitia_client.entities.request.route_schedule import RouteScheduleRequest from navitia_client.entities.response.route_schedule import RouteSchedule class RouteSchedulesApiClient(ApiBaseClient): """ A client class to interact with the Navitia API for fetching route schedules. + Uses the RouteScheduleRequest class to encapsulate query parameters. See https://doc.navitia.io/#route-schedules - - Methods - ------- - _get_route_schedule_object_from_response(response: Any) -> Sequence[RouteSchedule]: - A static method to transform raw API response data into a list of RouteSchedule objects. - - list_route_schedules_by_region_id_and_path( - region_id: str, - resource_path: str, - 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]: - Retrieves route schedules for a specified region and resource path from the Navitia API. - - list_route_schedules_by_coordinates( - 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]: - Retrieves route schedules for a specified set of coordinates from the Navitia API. """ @staticmethod @@ -84,53 +50,30 @@ def list_route_schedules_by_region_id_and_path( self, region_id: str, resource_path: str, - from_datetime: datetime = datetime.now(), - duration: int = 86400, - depth: int = 1, - count: int = 10, - start_page: int = 0, - items_per_schedule: int = 1, - forbidden_uris: Optional[Sequence[str]] = None, - data_freshness: str = "base_schedule", - disable_geojson: bool = False, - direction_type: str = "all", + request: RouteScheduleRequest, ) -> Sequence[RouteSchedule]: """ Retrieves route schedules for a specified region and resource path from the Navitia API. - Parameters: - region_id (str): The region ID. - resource_path (str): The resource path. - from_datetime (datetime, optional): The start datetime for the schedule. Defaults to datetime.now(). - duration (int, optional): The duration of the schedule in seconds. Defaults to 86400. - depth (int, optional): The depth of data to retrieve. Defaults to 1. - count (int, optional): Maximum number of results. Defaults to 10. - start_page (int, optional): The page number to start from. Defaults to 0. - items_per_schedule (int, optional): The number of items per schedule. Defaults to 1. - forbidden_uris (Optional[Sequence[str]], optional): Forbidden URIs. Defaults to None. - data_freshness (str, optional): The freshness of data to retrieve. Defaults to "base_schedule". - disable_geojson (bool, optional): Whether to disable GeoJSON in the response. Defaults to False. - direction_type (str, optional): The direction type. Defaults to "all". - - Returns: - Sequence[RouteSchedule]: A sequence of RouteSchedule objects. + Parameters + ---------- + region_id : str + The region ID. + resource_path : str + The resource path. + request : RouteScheduleRequest + The request object containing query parameters such as from_datetime, + duration, depth, count, start_page, items_per_schedule, forbidden_uris, + data_freshness, disable_geojson, and direction_type. + + Returns + ------- + Sequence[RouteSchedule] + A sequence of RouteSchedule objects. """ request_url = f"{self.base_navitia_url}/coverage/{region_id}/{resource_path}/route_schedules" - filters = { - "from_datetime": from_datetime, - "duration": duration, - "depth": depth, - "count": count, - "start_page": start_page, - "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) + return self._get_routes_nearby(request_url, request.to_filters()) def list_route_schedules_by_coordinates( self, @@ -138,52 +81,31 @@ def list_route_schedules_by_coordinates( region_lat: float, lon: float, lat: float, - from_datetime: datetime = datetime.now(), - duration: int = 86400, - depth: int = 1, - count: int = 10, - start_page: int = 0, - items_per_schedule: int = 1, - forbidden_uris: Optional[Sequence[str]] = None, - data_freshness: str = "base_schedule", - disable_geojson: bool = False, - direction_type: str = "all", + request: RouteScheduleRequest, ) -> Sequence[RouteSchedule]: """ Retrieves route schedules for a specified set of coordinates from the Navitia API. - Parameters: - region_lon (float): The longitude of the region. - region_lat (float): The latitude of the region. - lon (float): The longitude of the coordinates. - lat (float): The latitude of the coordinates. - from_datetime (datetime, optional): The start datetime for the schedule. Defaults to datetime.now(). - duration (int, optional): The duration of the schedule in seconds. Defaults to 86400. - depth (int, optional): The depth of data to retrieve. Defaults to 1. - count (int, optional): Maximum number of results. Defaults to 10. - start_page (int, optional): The page number to start from. Defaults to 0. - items_per_schedule (int, optional): The number of items per schedule. Defaults to 1. - forbidden_uris (Optional[Sequence[str]], optional): Forbidden URIs. Defaults to None. - data_freshness (str, optional): The freshness of data to retrieve. Defaults to "base_schedule". - disable_geojson (bool, optional): Whether to disable GeoJSON in the response. Defaults to False. - direction_type (str, optional): The direction type. Defaults to "all". - - Returns: - Sequence[RouteSchedule]: A sequence of RouteSchedule objects. + Parameters + ---------- + region_lon : float + The longitude of the region. + region_lat : float + The latitude of the region. + lon : float + The longitude of the coordinates. + lat : float + The latitude of the coordinates. + request : RouteScheduleRequest + The request object containing query parameters such as from_datetime, + duration, depth, count, start_page, items_per_schedule, forbidden_uris, + data_freshness, disable_geojson, and direction_type. + + Returns + ------- + Sequence[RouteSchedule] + A sequence of RouteSchedule objects. """ request_url = f"{self.base_navitia_url}/coverage/{region_lon};{region_lat}/coords/{lon};{lat}/route_schedules" - filters = { - "from_datetime": from_datetime, - "duration": duration, - "depth": depth, - "count": count, - "start_page": start_page, - "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) + return self._get_routes_nearby(request_url, request.to_filters()) diff --git a/navitia_client/client/apis/stop_schedules_apis.py b/navitia_client/client/apis/stop_schedules_apis.py index b462476..5c8f59a 100644 --- a/navitia_client/client/apis/stop_schedules_apis.py +++ b/navitia_client/client/apis/stop_schedules_apis.py @@ -1,6 +1,6 @@ -from datetime import datetime -from typing import Any, Optional, Sequence, Tuple +from typing import Any, Sequence, Tuple from navitia_client.client.apis.api_base_client import ApiBaseClient +from navitia_client.entities.request.stop_schedule import StopScheduleRequest from navitia_client.entities.response import Pagination from navitia_client.entities.response.stop_schedule import StopSchedule @@ -8,43 +8,9 @@ class StopSchedulesApiClient(ApiBaseClient): """ A client class to interact with the Navitia API for fetching stop schedules. + Uses the StopScheduleRequest class to encapsulate query parameters. See https://doc.navitia.io/#stop-schedules - - Methods - ------- - _get_stop_schedule_objects_from_response(response: Any) -> Sequence[StopSchedule]: - A static method to transform raw API response data into a list of StopSchedule objects. - - list_stop_schedules_by_coordinates( - 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 = "realtime", - disable_geojson: bool = False, - direction_type: str = "all", - ) -> Tuple[Sequence[StopSchedule], Pagination]: - Retrieves stop schedules for a specified set of coordinates from the Navitia API. - - list_stop_schedules_by_region_id_and_path( - region_id: str, - resource_path: str, - 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 = "realtime", - disable_geojson: bool = False, - direction_type: str = "all", - ) -> Tuple[Sequence[StopSchedule], Pagination]: - Retrieves stop schedules for a specified region and resource path from the Navitia API. """ @staticmethod @@ -90,104 +56,60 @@ def list_stop_schedules_by_coordinates( region_lat: float, lon: float, lat: float, - from_datetime: datetime = datetime.now(), - duration: int = 86400, - depth: int = 1, - count: int = 10, - start_page: int = 0, - items_per_schedule: int = 1, - forbidden_uris: Optional[Sequence[str]] = None, - data_freshness: str = "realtime", - disable_geojson: bool = False, - direction_type: str = "all", + request: StopScheduleRequest, ) -> Tuple[Sequence[StopSchedule], Pagination]: """ Retrieves stop schedules for a specified set of coordinates from the Navitia API. - Parameters: - region_lon (float): The longitude of the region. - region_lat (float): The latitude of the region. - lon (float): The longitude of the coordinates. - lat (float): The latitude of the coordinates. - from_datetime (datetime, optional): The start datetime for the schedule. Defaults to datetime.now(). - duration (int, optional): The duration of the schedule in seconds. Defaults to 86400. - depth (int, optional): The depth of data to retrieve. Defaults to 1. - count (int, optional): Maximum number of results. Defaults to 10. - start_page (int, optional): The page number to start from. Defaults to 0. - items_per_schedule (int, optional): The number of items per schedule. Defaults to 1. - forbidden_uris (Optional[Sequence[str]], optional): Forbidden URIs. Defaults to None. - data_freshness (str, optional): The freshness of data to retrieve. Defaults to "realtime". - disable_geojson (bool, optional): Whether to disable GeoJSON in the response. Defaults to False. - direction_type (str, optional): The direction type. Defaults to "all". - - Returns: - Tuple[Sequence[StopSchedule], Pagination]: A tuple containing a sequence of StopSchedule objects and Pagination object. + Parameters + ---------- + region_lon : float + The longitude of the region. + region_lat : float + The latitude of the region. + lon : float + The longitude of the coordinates. + lat : float + The latitude of the coordinates. + request : StopScheduleRequest + The request object containing query parameters such as from_datetime, + duration, depth, count, start_page, items_per_schedule, forbidden_uris, + data_freshness, disable_geojson, and direction_type. + + Returns + ------- + Tuple[Sequence[StopSchedule], Pagination] + A tuple containing a sequence of StopSchedule objects and Pagination object. """ request_url = f"{self.base_navitia_url}/coverage/{region_lon};{region_lat}/coords/{lon};{lat}/stop_schedules" - filters = { - "from_datetime": from_datetime, - "duration": duration, - "depth": depth, - "count": count, - "start_page": start_page, - "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_stop_schedules(request_url, filters) + return self._get_stop_schedules(request_url, request.to_filters()) def list_stop_schedules_by_region_id_and_path( self, region_id: str, resource_path: str, - from_datetime: datetime = datetime.now(), - duration: int = 86400, - depth: int = 1, - count: int = 10, - start_page: int = 0, - items_per_schedule: int = 1, - forbidden_uris: Optional[Sequence[str]] = None, - data_freshness: str = "realtime", - disable_geojson: bool = False, - direction_type: str = "all", + request: StopScheduleRequest, ) -> Tuple[Sequence[StopSchedule], Pagination]: """ Retrieves stop schedules for a specified region and resource path from the Navitia API. - Parameters: - region_id (str): The region ID. - resource_path (str): The resource path. - from_datetime (datetime, optional): The start datetime for the schedule. Defaults to datetime.now(). - duration (int, optional): The duration of the schedule in seconds. Defaults to 86400. - depth (int, optional): The depth of data to retrieve. Defaults to 1. - count (int, optional): Maximum number of results. Defaults to 10. - start_page (int, optional): The page number to start from. Defaults to 0. - items_per_schedule (int, optional): The number of items per schedule. Defaults to 1. - forbidden_uris (Optional[Sequence[str]], optional): Forbidden URIs. Defaults to None. - data_freshness (str, optional): The freshness of data to retrieve. Defaults to "realtime". - disable_geojson (bool, optional): Whether to disable GeoJSON in the response. Defaults to False. - direction_type (str, optional): The direction type. Defaults to "all". - - Returns: - Tuple[Sequence[StopSchedule], Pagination]: A tuple containing a sequence of StopSchedule objects and Pagination object. + Parameters + ---------- + region_id : str + The region ID. + resource_path : str + The resource path. + request : StopScheduleRequest + The request object containing query parameters such as from_datetime, + duration, depth, count, start_page, items_per_schedule, forbidden_uris, + data_freshness, disable_geojson, and direction_type. + + Returns + ------- + Tuple[Sequence[StopSchedule], Pagination] + A tuple containing a sequence of StopSchedule objects and Pagination object. """ request_url = f"{self.base_navitia_url}/coverage/{region_id}/{resource_path}/stop_schedules" - filters = { - "from_datetime": from_datetime, - "duration": duration, - "depth": depth, - "count": count, - "start_page": start_page, - "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_stop_schedules(request_url, filters) + return self._get_stop_schedules(request_url, request.to_filters()) diff --git a/navitia_client/client/apis/terminus_schedules_apis.py b/navitia_client/client/apis/terminus_schedules_apis.py index d903ea2..696cd3a 100644 --- a/navitia_client/client/apis/terminus_schedules_apis.py +++ b/navitia_client/client/apis/terminus_schedules_apis.py @@ -1,6 +1,6 @@ -from datetime import datetime -from typing import Any, Optional, Sequence, Tuple +from typing import Any, Sequence, Tuple from navitia_client.client.apis.api_base_client import ApiBaseClient +from navitia_client.entities.request.terminus_schedule import TerminusScheduleRequest from navitia_client.entities.response import Pagination from navitia_client.entities.response.stop_schedule import TerminusSchedule @@ -8,43 +8,9 @@ class TerminusSchedulesApiClient(ApiBaseClient): """ A client class to interact with the Navitia API for fetching terminus schedules. + Uses the TerminusScheduleRequest class to encapsulate query parameters. See https://doc.navitia.io/#terminus-schedules - - Methods - _get_terminus_schedule_objects_from_response(response: Any) -> Sequence[TerminusSchedule]: - ------- - A static method to transform raw API response data into a list of TerminusSchedule objects. - - list_terminus_schedules_by_region_id_and_path( - region_id: str, - resource_path: str, - 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 = "realtime", - disable_geojson: bool = False, - direction_type: str = "all", - ) -> Tuple[Sequence[TerminusSchedule], Pagination]: - Retrieves terminus schedules for a specified region and resource path from the Navitia API. - - list_terminus_schedules_by_coordinates( - 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 = "realtime", - disable_geojson: bool = False, - direction_type: str = "all", - ) -> Tuple[Sequence[TerminusSchedule], Pagination]: - Retrieves terminus schedules for a specified set of coordinates from the Navitia API. """ @staticmethod @@ -92,53 +58,30 @@ def list_terminus_schedules_by_region_id_and_path( self, region_id: str, resource_path: str, - from_datetime: datetime = datetime.now(), - duration: int = 86400, - depth: int = 1, - count: int = 10, - start_page: int = 0, - items_per_schedule: int = 1, - forbidden_uris: Optional[Sequence[str]] = None, - data_freshness: str = "realtime", - disable_geojson: bool = False, - direction_type: str = "all", + request: TerminusScheduleRequest, ) -> Tuple[Sequence[TerminusSchedule], Pagination]: """ Retrieves terminus schedules for a specified region and resource path from the Navitia API. - Parameters: - region_id (str): The region ID. - resource_path (str): The resource path. - from_datetime (datetime, optional): The start datetime for the schedule. Defaults to datetime.now(). - duration (int, optional): The duration of the schedule in seconds. Defaults to 86400. - depth (int, optional): The depth of data to retrieve. Defaults to 1. - count (int, optional): Maximum number of results. Defaults to 10. - start_page (int, optional): The page number to start from. Defaults to 0. - items_per_schedule (int, optional): The number of items per schedule. Defaults to 1. - forbidden_uris (Optional[Sequence[str]], optional): Forbidden URIs. Defaults to None. - data_freshness (str, optional): The freshness of data to retrieve. Defaults to "realtime". - disable_geojson (bool, optional): Whether to disable GeoJSON in the response. Defaults to False. - direction_type (str, optional): The direction type. Defaults to "all". - - Returns: - Tuple[Sequence[TerminusSchedule], Pagination]: A tuple containing a sequence of TerminusSchedule objects and Pagination object. + Parameters + ---------- + region_id : str + The region ID. + resource_path : str + The resource path. + request : TerminusScheduleRequest + The request object containing query parameters such as from_datetime, + duration, depth, count, start_page, items_per_schedule, forbidden_uris, + data_freshness, disable_geojson, and direction_type. + + Returns + ------- + Tuple[Sequence[TerminusSchedule], Pagination] + A tuple containing a sequence of TerminusSchedule objects and Pagination object. """ request_url = f"{self.base_navitia_url}/coverage/{region_id}/{resource_path}/terminus_schedules" - filters = { - "from_datetime": from_datetime, - "duration": duration, - "depth": depth, - "count": count, - "start_page": start_page, - "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_stop_schedules(request_url, filters) + return self._get_stop_schedules(request_url, request.to_filters()) def list_terminus_schedules_by_coordinates( self, @@ -146,53 +89,31 @@ def list_terminus_schedules_by_coordinates( region_lat: float, lon: float, lat: float, - from_datetime: datetime = datetime.now(), - duration: int = 86400, - depth: int = 1, - count: int = 10, - start_page: int = 0, - items_per_schedule: int = 1, - forbidden_uris: Optional[Sequence[str]] = None, - data_freshness: str = "realtime", - disable_geojson: bool = False, - direction_type: str = "all", + request: TerminusScheduleRequest, ) -> Tuple[Sequence[TerminusSchedule], Pagination]: """ Retrieves terminus schedules for a specified set of coordinates from the Navitia API. - Parameters: - region_lon (float): The longitude of the region. - region_lat (float): The latitude of the region. - lon (float): The longitude of the coordinates. - lat (float): The latitude of the coordinates. - from_datetime (datetime, optional): The start datetime for the schedule. Defaults to datetime.now(). - duration (int, optional): The duration of the schedule in seconds. Defaults to 86400. - depth (int, optional): The depth of data to retrieve. Defaults to 1. - count (int, optional): Maximum number of results. Defaults to 10. - start_page (int, optional): The page number to start from. Defaults to 0. - items_per_schedule (int, optional): The number of items per schedule. Defaults to 1. - forbidden_uris (Optional[Sequence[str]], optional): Forbidden URIs. Defaults to None. - data_freshness (str, optional): The freshness of data to retrieve. Defaults to "realtime". - disable_geojson (bool, optional): Whether to disable GeoJSON in the response. Defaults to False. - direction_type (str, optional): The direction type. Defaults to "all". - - Returns: - Tuple[Sequence[TerminusSchedule], Pagination]: A tuple containing a sequence of TerminusSchedule objects and Pagination object. + Parameters + ---------- + region_lon : float + The longitude of the region. + region_lat : float + The latitude of the region. + lon : float + The longitude of the coordinates. + lat : float + The latitude of the coordinates. + request : TerminusScheduleRequest + The request object containing query parameters such as from_datetime, + duration, depth, count, start_page, items_per_schedule, forbidden_uris, + data_freshness, disable_geojson, and direction_type. + + Returns + ------- + Tuple[Sequence[TerminusSchedule], Pagination] + A tuple containing a sequence of TerminusSchedule objects and Pagination object. """ - request_url = f"{self.base_navitia_url}/coverage/{region_lon};{region_lat}/coords/{lon};{lat}/terminus_schedules" - filters = { - "from_datetime": from_datetime, - "duration": duration, - "depth": depth, - "count": count, - "start_page": start_page, - "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_stop_schedules(request_url, filters) + return self._get_stop_schedules(request_url, request.to_filters()) diff --git a/navitia_client/client/apis/traffic_report_apis.py b/navitia_client/client/apis/traffic_report_apis.py index 448d866..4e4a44a 100644 --- a/navitia_client/client/apis/traffic_report_apis.py +++ b/navitia_client/client/apis/traffic_report_apis.py @@ -1,6 +1,6 @@ -from datetime import datetime from typing import Optional, Sequence, Tuple from navitia_client.client.apis.api_base_client import ApiBaseClient +from navitia_client.entities.request.traffic_report import TrafficReportRequest from navitia_client.entities.response.disruption import Disruption from navitia_client.entities.response import Pagination from navitia_client.entities.response.traffic_report import TrafficReport @@ -22,12 +22,7 @@ class TrafficReportsApiClient(ApiBaseClient): list_traffic_reports( region_id: Optional[str] = None, resource_path: Optional[str] = None, - since: Optional[datetime] = None, - until: Optional[datetime] = None, - count: int = 25, - depth: int = 1, - forbidden_uris: Optional[Sequence[str]] = None, - disable_geojson: bool = False, + request: TrafficReportRequest = TrafficReportRequest(), ) -> Tuple[Sequence[Disruption], Sequence[TrafficReport], Pagination]: Retrieves traffic reports for a specified region and resource path from the Navitia API. """ @@ -60,12 +55,7 @@ def list_traffic_reports( self, region_id: Optional[str] = None, resource_path: Optional[str] = None, - since: Optional[datetime] = None, - until: Optional[datetime] = None, - count: int = 25, - depth: int = 1, - forbidden_uris: Optional[Sequence[str]] = None, - disable_geojson: bool = False, + request: TrafficReportRequest = TrafficReportRequest(), ) -> Tuple[Sequence[Disruption], Sequence[TrafficReport], Pagination]: """ Retrieves traffic reports for a specified region and resource path from the Navitia API. @@ -73,12 +63,7 @@ def list_traffic_reports( Parameters: region_id (Optional[str]): The region ID. resource_path (Optional[str]): The resource path. - since (Optional[datetime]): The start datetime for the reports. - until (Optional[datetime]): The end datetime for the reports. - count (int): The number of reports to retrieve. Defaults to 25. - depth (int): The depth of data to retrieve. Defaults to 1. - forbidden_uris (Optional[Sequence[str]]): Forbidden URIs. - disable_geojson (bool): Whether to disable GeoJSON in the response. Defaults to False. + request (TrafficReportRequest): The request object containing query parameters. Returns: Tuple[Sequence[Disruption], Sequence[TrafficReport], Pagination]: A tuple containing sequences of Disruption and TrafficReport objects, and Pagination object. @@ -90,17 +75,4 @@ def list_traffic_reports( f"{self.base_navitia_url}/coverage/{region_id}/traffic_reports" ) - filters = { - "count": count, - "depth": depth, - "disable_geojson": disable_geojson, - "forbidden_uris[]": forbidden_uris, - } - - if since: - filters["since"] = since.isoformat() - - if until: - filters["until"] = until.isoformat() - - return self._get_traffic_reports(request_url, filters) + return self._get_traffic_reports(request_url, request.to_filters()) diff --git a/navitia_client/entities/request/__init__.py b/navitia_client/entities/request/__init__.py new file mode 100644 index 0000000..8192f72 --- /dev/null +++ b/navitia_client/entities/request/__init__.py @@ -0,0 +1,45 @@ +from navitia_client.entities.request.arrival import ArrivalRequest +from navitia_client.entities.request.base_entity_request import BaseEntityRequest +from navitia_client.entities.request.contributor import ContributorRequest +from navitia_client.entities.request.coverage import CoverageRequest +from navitia_client.entities.request.dataset import DatasetRequest +from navitia_client.entities.request.departure import DepartureRequest +from navitia_client.entities.request.equipment_report import EquipmentReportRequest +from navitia_client.entities.request.freefloatings_nearby import ( + FreefloatingsNearbyRequest, +) +from navitia_client.entities.request.inverted_geocoding import InvertedGeocodingRequest +from navitia_client.entities.request.isochrone import IsochroneRequest +from navitia_client.entities.request.journey import JourneyRequest +from navitia_client.entities.request.line_report import LineReportRequest +from navitia_client.entities.request.place import PlaceRequest +from navitia_client.entities.request.places_nearby import PlacesNearbyRequest +from navitia_client.entities.request.public_transport_object import ( + PublicTransportObjectRequest, +) +from navitia_client.entities.request.route_schedule import RouteScheduleRequest +from navitia_client.entities.request.stop_schedule import StopScheduleRequest +from navitia_client.entities.request.terminus_schedule import TerminusScheduleRequest +from navitia_client.entities.request.traffic_report import TrafficReportRequest + +__all__ = [ + "ArrivalRequest", + "BaseEntityRequest", + "ContributorRequest", + "CoverageRequest", + "DatasetRequest", + "DepartureRequest", + "EquipmentReportRequest", + "FreefloatingsNearbyRequest", + "InvertedGeocodingRequest", + "IsochroneRequest", + "JourneyRequest", + "LineReportRequest", + "PlaceRequest", + "PlacesNearbyRequest", + "PublicTransportObjectRequest", + "RouteScheduleRequest", + "StopScheduleRequest", + "TerminusScheduleRequest", + "TrafficReportRequest", +] diff --git a/navitia_client/entities/request/arrival.py b/navitia_client/entities/request/arrival.py new file mode 100644 index 0000000..ba6f72b --- /dev/null +++ b/navitia_client/entities/request/arrival.py @@ -0,0 +1,62 @@ +from dataclasses import dataclass, field +from datetime import datetime +from typing import Dict, Any, Optional, Sequence + +from .base_entity_request import BaseEntityRequest + + +@dataclass +class ArrivalRequest(BaseEntityRequest): + """ + Request class for Arrival entities. + + Attributes + ---------- + from_datetime : datetime, optional + The starting datetime for fetching arrivals (default is current datetime). + duration : int, optional + The duration in seconds for which to fetch arrivals (default is 86400 seconds). + depth : int, optional + The depth of the search (default is 1). + forbidden_uris : Optional[Sequence[str]], optional + A list of URIs to exclude from the search (default is None). + data_freshness : str, optional + The freshness of the data to fetch, either "realtime" or "base_schedule" (default is "realtime"). + disable_geojson : bool, optional + Whether to disable geoJSON in the response (default is False). + direction_type : str, optional + The direction type of the arrivals to fetch, e.g., "all", "forward", "backward" (default is "all"). + + """ + + from_datetime: datetime = field(default_factory=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" + + def to_filters(self) -> Dict[str, Any]: + """ + Converts the ArrivalRequest instance into a dictionary of filters. + + Returns: + Dict[str, Any]: A dictionary representation of the request filters. + """ + filters = { + "from_datetime": self.from_datetime, + "duration": self.duration, + "depth": self.depth, + "count": self.count, + "start_page": self.start_page, + "data_freshness": self.data_freshness, + "direction_type": self.direction_type, + } + + if self.disable_geojson: + filters["disable_geojson"] = self.disable_geojson + if self.forbidden_uris: + filters["forbidden_uris[]"] = self.forbidden_uris + + return filters diff --git a/navitia_client/entities/request/base_entity_request.py b/navitia_client/entities/request/base_entity_request.py new file mode 100644 index 0000000..3f7a5a4 --- /dev/null +++ b/navitia_client/entities/request/base_entity_request.py @@ -0,0 +1,52 @@ +from abc import ABC, abstractmethod +from typing import Any, Dict, Optional + + +class BaseEntityRequest(ABC): + """ + Base class for entity requests. + + count : int, optional + Maximum number of results (default is 10). + start_page : int, optional + The page number to start from (default is 0). + + """ + + count: int = 10 + start_page: int = 0 + + @abstractmethod + def to_filters(self) -> Dict[str, Any]: + pass + + +class BasePTEntityRequest(BaseEntityRequest): + """ + Base class for public transport entity requests. + Inherits from BaseEntityRequest. + depth : int, optional + Search depth (default is 1). + disable_geojson : bool, optional + Whether to disable GeoJSON in the response (default is False). + odt : str, optional + ODT type filter (default is "all"). + distance : int, optional + Maximum search distance (default is 200). + headsign : Optional[str], optional + Line headsign. + since : Optional[str], optional + To be used only on "vehicle_journeys" and "disruptions" collection, to filter on a period. + until : Optional[str], optional + To be used only on "vehicle_journeys" and "disruptions" collection, to filter on a period. + + """ + + depth: int = 1 + odt: str = "all" + distance: int = 200 + headsign: Optional[str] = None + since: Optional[str] = None + until: Optional[str] = None + disable_geojson: bool = False + disable_disruption: bool = False diff --git a/navitia_client/entities/request/contributor.py b/navitia_client/entities/request/contributor.py new file mode 100644 index 0000000..4cad8d3 --- /dev/null +++ b/navitia_client/entities/request/contributor.py @@ -0,0 +1,28 @@ +from dataclasses import dataclass +from typing import Dict, Any + +from .base_entity_request import BaseEntityRequest + + +@dataclass +class ContributorRequest(BaseEntityRequest): + """ + Request class for Contributor queries. + + This class only uses the pagination parameters (count, start_page) + inherited from BaseEntityRequest. + """ + + def to_filters(self) -> Dict[str, Any]: + """ + Converts the ContributorRequest instance into a dictionary of filters. + + Returns + ------- + Dict[str, Any] + A dictionary representation of the request filters. + """ + return { + "count": self.count, + "start_page": self.start_page, + } diff --git a/navitia_client/entities/request/coverage.py b/navitia_client/entities/request/coverage.py new file mode 100644 index 0000000..05f938b --- /dev/null +++ b/navitia_client/entities/request/coverage.py @@ -0,0 +1,28 @@ +from dataclasses import dataclass +from typing import Dict, Any + +from .base_entity_request import BaseEntityRequest + + +@dataclass +class CoverageRequest(BaseEntityRequest): + """ + Request class for Coverage queries. + + This class only uses the pagination parameters (count, start_page) + inherited from BaseEntityRequest. + """ + + def to_filters(self) -> Dict[str, Any]: + """ + Converts the CoverageRequest instance into a dictionary of filters. + + Returns + ------- + Dict[str, Any] + A dictionary representation of the request filters. + """ + return { + "count": self.count, + "start_page": self.start_page, + } diff --git a/navitia_client/entities/request/dataset.py b/navitia_client/entities/request/dataset.py new file mode 100644 index 0000000..bf41daf --- /dev/null +++ b/navitia_client/entities/request/dataset.py @@ -0,0 +1,28 @@ +from dataclasses import dataclass +from typing import Dict, Any + +from .base_entity_request import BaseEntityRequest + + +@dataclass +class DatasetRequest(BaseEntityRequest): + """ + Request class for Dataset queries. + + This class only uses the pagination parameters (count, start_page) + inherited from BaseEntityRequest. + """ + + def to_filters(self) -> Dict[str, Any]: + """ + Converts the DatasetRequest instance into a dictionary of filters. + + Returns + ------- + Dict[str, Any] + A dictionary representation of the request filters. + """ + return { + "count": self.count, + "start_page": self.start_page, + } diff --git a/navitia_client/entities/request/departure.py b/navitia_client/entities/request/departure.py new file mode 100644 index 0000000..0f786b6 --- /dev/null +++ b/navitia_client/entities/request/departure.py @@ -0,0 +1,67 @@ +from dataclasses import dataclass, field +from datetime import datetime +from typing import Dict, Any, Optional, Sequence + +from .base_entity_request import BaseEntityRequest + + +@dataclass +class DepartureRequest(BaseEntityRequest): + """ + Request class for Departure entities. + + Attributes + ---------- + from_datetime : datetime, optional + The starting datetime for fetching departures (default is current datetime). + duration : int, optional + The duration in seconds for which to fetch departures (default is 86400 seconds). + depth : int, optional + The depth of the search (default is 1). + count : int, optional + Maximum number of results (default is 10, inherited from BaseEntityRequest). + start_page : int, optional + The page number to start from (default is 0, inherited from BaseEntityRequest). + forbidden_uris : Optional[Sequence[str]], optional + A list of URIs to exclude from the search (default is None). + data_freshness : str, optional + The freshness of the data to fetch, either "realtime" or "base_schedule" (default is "realtime"). + disable_geojson : bool, optional + Whether to disable geoJSON in the response (default is False). + direction_type : str, optional + The direction type of the departures to fetch, e.g., "all", "forward", "backward" (default is "all"). + """ + + from_datetime: datetime = field(default_factory=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" + + def to_filters(self) -> Dict[str, Any]: + """ + Converts the DepartureRequest instance into a dictionary of filters. + + Returns + ------- + Dict[str, Any] + A dictionary representation of the request filters. + """ + filters = { + "from_datetime": self.from_datetime, + "duration": self.duration, + "depth": self.depth, + "count": self.count, + "start_page": self.start_page, + "data_freshness": self.data_freshness, + "direction_type": self.direction_type, + } + + if self.disable_geojson: + filters["disable_geojson"] = self.disable_geojson + if self.forbidden_uris: + filters["forbidden_uris[]"] = self.forbidden_uris + + return filters diff --git a/navitia_client/entities/request/equipment_report.py b/navitia_client/entities/request/equipment_report.py new file mode 100644 index 0000000..c25c956 --- /dev/null +++ b/navitia_client/entities/request/equipment_report.py @@ -0,0 +1,46 @@ +from dataclasses import dataclass +from typing import Dict, Any, Optional, Sequence + +from .base_entity_request import BaseEntityRequest + + +@dataclass +class EquipmentReportRequest(BaseEntityRequest): + """ + Request class for Equipment Report queries. + + Attributes + ---------- + depth : int, optional + Json response depth (default is 1). + filter : Optional[str], optional + A filter to refine your request (default is None). + forbidden_uris : Optional[Sequence[str]], optional + If you want to avoid lines, modes, networks, etc (default is None). + """ + + depth: int = 1 + filter: Optional[str] = None + forbidden_uris: Optional[Sequence[str]] = None + + def to_filters(self) -> Dict[str, Any]: + """ + Converts the EquipmentReportRequest instance into a dictionary of filters. + + Returns + ------- + Dict[str, Any] + A dictionary representation of the request filters. + """ + filters: Dict[str, Any] = { + "count": self.count, + "depth": self.depth, + "start_page": self.start_page, + } + + if self.filter: + filters["filter"] = self.filter + if self.forbidden_uris: + filters["forbidden_uris[]"] = self.forbidden_uris + + return filters diff --git a/navitia_client/entities/request/freefloatings_nearby.py b/navitia_client/entities/request/freefloatings_nearby.py new file mode 100644 index 0000000..1c00827 --- /dev/null +++ b/navitia_client/entities/request/freefloatings_nearby.py @@ -0,0 +1,40 @@ +from dataclasses import dataclass +from typing import Dict, Any, Optional, Sequence + +from .base_entity_request import BaseEntityRequest + + +@dataclass +class FreefloatingsNearbyRequest(BaseEntityRequest): + """ + Request class for Freefloatings Nearby queries. + + Attributes + ---------- + distance : int, optional + Search radius in meters (default is 500). + type : Optional[Sequence[str]], optional + The type of shared mobility vehicles to return (e.g., bike, scooter, car) (default is None). + """ + + distance: int = 500 + type: Optional[Sequence[str]] = None + + def to_filters(self) -> Dict[str, Any]: + """ + Converts the FreefloatingsNearbyRequest instance into a dictionary of filters. + + Returns + ------- + Dict[str, Any] + A dictionary representation of the request filters. + """ + filters: Dict[str, Any] = { + "distance": self.distance, + "count": self.count, + } + + if self.type: + filters["type[]"] = self.type + + return filters diff --git a/navitia_client/entities/request/inverted_geocoding.py b/navitia_client/entities/request/inverted_geocoding.py new file mode 100644 index 0000000..b46c89f --- /dev/null +++ b/navitia_client/entities/request/inverted_geocoding.py @@ -0,0 +1,29 @@ +from dataclasses import dataclass +from typing import Dict, Any + +from .base_entity_request import BaseEntityRequest + + +@dataclass +class InvertedGeocodingRequest(BaseEntityRequest): + """ + Request class for Inverted Geocoding queries. + + Note: Inverted Geocoding API doesn't accept any additional filters beyond base parameters. + This class inherits count and start_page from BaseEntityRequest. + """ + + def to_filters(self) -> Dict[str, Any]: + """ + Converts the InvertedGeocodingRequest instance into a dictionary of filters. + + Returns + ------- + Dict[str, Any] + A dictionary representation of the request filters. + """ + filters = { + "count": self.count, + } + + return filters diff --git a/navitia_client/entities/request/isochrone.py b/navitia_client/entities/request/isochrone.py new file mode 100644 index 0000000..7ac54f4 --- /dev/null +++ b/navitia_client/entities/request/isochrone.py @@ -0,0 +1,75 @@ +from dataclasses import dataclass, field +from datetime import datetime +from typing import Dict, Any, Optional, Sequence + +from .base_entity_request import BaseEntityRequest + + +@dataclass +class IsochroneRequest(BaseEntityRequest): + """ + Request class for Isochrone queries. + + Attributes + ---------- + from_ : str + The starting point for the isochrone calculation (required). + start_datetime : datetime, optional + The starting date and time for the isochrone calculation (default is datetime.now()). + boundary_duration : Sequence[int], optional + List of durations in seconds defining the isochrones boundaries (default is empty list). + to : Optional[str], optional + The ending point for the isochrone calculation (default is None). + first_section_mode : Optional[Sequence[str]], optional + Modes of transportation for the first section of the journey (default is None). + last_section_mode : Optional[Sequence[str]], optional + Modes of transportation for the last section of the journey (default is None). + min_duration : Optional[int], optional + The minimum duration for the isochrone calculation (default is None). + max_duration : Optional[int], optional + The maximum duration for the isochrone calculation (default is None). + """ + + from_: str + start_datetime: datetime = field(default_factory=datetime.now) + boundary_duration: Sequence[int] = field(default_factory=list) + to: Optional[str] = None + first_section_mode: Optional[Sequence[str]] = None + last_section_mode: Optional[Sequence[str]] = None + min_duration: Optional[int] = None + max_duration: Optional[int] = None + + def to_filters(self) -> Dict[str, Any]: + """ + Converts the IsochroneRequest instance into a dictionary of filters. + + Returns + ------- + Dict[str, Any] + A dictionary representation of the request filters. + """ + filters: Dict[str, Any] = { + "datetime": self.start_datetime.isoformat(), + "boundary_duration[]": self.boundary_duration, + "from": self.from_, + } + + if self.to: + filters["to"] = self.to + + if self.min_duration is not None: + filters["min_duration"] = self.min_duration + + if self.first_section_mode: + filters["first_section_mode[]"] = self.first_section_mode + + if self.last_section_mode: + filters["last_section_mode[]"] = self.last_section_mode + + if self.max_duration is not None: + filters["max_duration"] = self.max_duration + # Special logic: if max_duration is set and boundary_duration is empty, remove min_duration + if len(self.boundary_duration) == 0: + filters.pop("min_duration", None) + + return filters diff --git a/navitia_client/entities/request/journey.py b/navitia_client/entities/request/journey.py new file mode 100644 index 0000000..d9b2aa0 --- /dev/null +++ b/navitia_client/entities/request/journey.py @@ -0,0 +1,215 @@ +from dataclasses import dataclass, field +from datetime import datetime +from typing import Dict, Any, Optional, Sequence + +from navitia_client.entities.response import ParkMode +from .base_entity_request import BaseEntityRequest + + +@dataclass +class JourneyRequest(BaseEntityRequest): + """ + Request class for Journey queries. + + Attributes + ---------- + from_ : Optional[str], optional + The starting point for the journey (default is None). + to_ : Optional[str], optional + The ending point for the journey (default is None). + datetime_ : datetime, optional + The date and time for the journey calculation (default is datetime.now()). + datetime_represents : str, optional + Represents whether the datetime is for departure or arrival (default is "departure"). + traveler_type : str, optional + The type of traveler (default is "standard"). + data_freshness : str, optional + The freshness of the data, can be "realtime" or "base_schedule" (default is "realtime"). + forbidden_uris : Optional[Sequence[str]], optional + A list of URIs that are forbidden in the journey calculation (default is None). + allowed_id : Optional[Sequence[str]], optional + A list of allowed IDs for the journey calculation (default is None). + first_section_mode : Optional[Sequence[str]], optional + Modes of transportation for the first section of the journey (default is None). + last_section_mode : Optional[Sequence[str]], optional + Modes of transportation for the last section of the journey (default is None). + language : str, optional + The language for the journey results (default is "en-GB"). + depth : int, optional + The depth of the journey search (default is 1). + max_duration_to_pt : int, optional + Maximum duration to public transportation in seconds (default is 30 * 60). + walking_speed : float, optional + Walking speed in meters per second (default is 1.12). + bike_speed : float, optional + Bike speed in meters per second (default is 4.1). + bss_speed : float, optional + Bike-sharing speed in meters per second (default is 4.1). + car_speed : float, optional + Car speed in meters per second (default is 16.8). + min_nb_journeys : int, optional + Minimum number of journeys to be returned (default is 1). + max_nb_journeys : int, optional + Maximum number of journeys to be returned (default is 1). + max_nb_transfers : int, optional + Maximum number of transfers allowed in the journey (default is 10). + min_nb_transfers : int, optional + Minimum number of transfers required in the journey (default is 0). + max_duration : int, optional + Maximum duration of the journey in seconds (default is 86400). + wheelchair : bool, optional + Whether the journey should be wheelchair accessible (default is False). + direct_path : str, optional + Preference for direct paths, can be "indifferent", "requested", or "forbidden" (default is "indifferent"). + direct_path_mode : Optional[Sequence[str]], optional + Modes of transportation for direct paths (default is None). + add_poi_infos : Optional[Sequence[str]], optional + Additional points of interest information to be included (default is None). + debug : bool, optional + Whether to include debug information in the response (default is False). + free_radius_from : int, optional + Free radius from the starting point in meters (default is 0). + free_radius_to : int, optional + Free radius to the ending point in meters (default is 0). + timeframe_duration : int, optional + Timeframe duration in seconds for the journey calculation (default is 0). + park_mode : ParkMode, optional + Parking mode for car-based journeys (default is ParkMode.NONE). + is_journey_schedules : bool, optional + Whether to return journey schedules instead of journeys (default is False). + bike_use_hills : Optional[float], optional + Valhalla parameter: preference for using hills when biking, from 0 (avoid) to 1 (prefer) (default is None). + walking_use_hills : Optional[float], optional + Valhalla parameter: preference for using hills when walking, from 0 (avoid) to 1 (prefer) (default is None). + bike_avoid_bad_surfaces : Optional[float], optional + Valhalla parameter: preference for avoiding bad surfaces when biking, from 0 (don't avoid) to 1 (strongly avoid) (default is None). + walking_step_penalty : Optional[float], optional + Valhalla parameter: penalty applied to steps when walking, in seconds (default is None). + bike_maneuver_penalty : Optional[float], optional + Valhalla parameter: penalty applied to maneuvers when biking, in seconds (default is None). + bike_use_living_streets : Optional[float], optional + Valhalla parameter: preference for using living streets when biking, from 0 (avoid) to 1 (prefer) (default is None). + """ + + from_: Optional[str] = None + to_: Optional[str] = None + datetime_: datetime = field(default_factory=datetime.now) + datetime_represents: str = "departure" + traveler_type: str = "standard" + data_freshness: str = "realtime" + forbidden_uris: Optional[Sequence[str]] = None + allowed_id: Optional[Sequence[str]] = None + first_section_mode: Optional[Sequence[str]] = None + last_section_mode: Optional[Sequence[str]] = None + language: str = "en-GB" + depth: int = 1 + max_duration_to_pt: int = 30 * 60 + walking_speed: float = 1.12 + bike_speed: float = 4.1 + bss_speed: float = 4.1 + car_speed: float = 16.8 + min_nb_journeys: int = 1 + max_nb_journeys: int = 1 + max_nb_transfers: int = 10 + min_nb_transfers: int = 0 + max_duration: int = 86400 + wheelchair: bool = False + direct_path: str = "indifferent" + direct_path_mode: Optional[Sequence[str]] = None + add_poi_infos: Optional[Sequence[str]] = None + debug: bool = False + free_radius_from: int = 0 + free_radius_to: int = 0 + timeframe_duration: int = 0 + park_mode: ParkMode = ParkMode.NONE + is_journey_schedules: bool = False + bike_use_hills: Optional[float] = None + walking_use_hills: Optional[float] = None + bike_avoid_bad_surfaces: Optional[float] = None + walking_step_penalty: Optional[float] = None + bike_maneuver_penalty: Optional[float] = None + bike_use_living_streets: Optional[float] = None + + def to_filters(self) -> Dict[str, Any]: + """ + Converts the JourneyRequest instance into a dictionary of filters. + + Returns + ------- + Dict[str, Any] + A dictionary representation of the request filters. + """ + filters = { + "datetime": self.datetime_.isoformat(), + "datetime_represents": self.datetime_represents, + "traveler_type": self.traveler_type, + "data_freshness": self.data_freshness, + "language": self.language, + "depth": self.depth, + "max_duration_to_pt": self.max_duration_to_pt, + "walking_speed": self.walking_speed, + "bike_speed": self.bike_speed, + "bss_speed": self.bss_speed, + "car_speed": self.car_speed, + "min_nb_journeys": self.min_nb_journeys, + "max_nb_journeys": self.max_nb_journeys, + "count": self.count, + "max_nb_transfers": self.max_nb_transfers, + "min_nb_transfers": self.min_nb_transfers, + "max_duration": self.max_duration, + "wheelchair": self.wheelchair, + "direct_path": self.direct_path, + "debug": self.debug, + "free_radius_from": self.free_radius_from, + "free_radius_to": self.free_radius_to, + "timeframe_duration": self.timeframe_duration, + "is_journey_schedules": self.is_journey_schedules, + "park_mode": self.park_mode.value, + } + + if self.from_: + filters["from"] = self.from_ + + if self.to_: + filters["to"] = self.to_ + + if self.bike_use_hills is not None: + filters["bike_use_hills"] = self.bike_use_hills + + if self.walking_use_hills is not None: + filters["walking_use_hills"] = self.walking_use_hills + + if self.bike_avoid_bad_surfaces is not None: + filters["bike_avoid_bad_surfaces"] = self.bike_avoid_bad_surfaces + + if self.walking_step_penalty is not None: + filters["walking_step_penalty"] = self.walking_step_penalty + + if self.bike_maneuver_penalty is not None: + filters["bike_maneuver_penalty"] = self.bike_maneuver_penalty + + if self.bike_use_living_streets is not None: + filters["bike_use_living_streets"] = self.bike_use_living_streets + + if self.forbidden_uris: + filters["forbidden_uris[]"] = self.forbidden_uris + + if self.allowed_id: + filters["allowed_id[]"] = self.allowed_id + + if self.first_section_mode: + filters["first_section_mode[]"] = self.first_section_mode + + if self.last_section_mode: + filters["last_section_mode[]"] = self.last_section_mode + + if self.add_poi_infos: + filters["add_poi_infos[]"] = self.add_poi_infos + + # Special logic: direct_path_mode defaults to first_section_mode if not provided + if any([self.direct_path_mode, self.first_section_mode]): + filters["direct_path_mode[]"] = ( + self.direct_path_mode or self.first_section_mode + ) + + return filters diff --git a/navitia_client/entities/request/line_report.py b/navitia_client/entities/request/line_report.py new file mode 100644 index 0000000..5bf938d --- /dev/null +++ b/navitia_client/entities/request/line_report.py @@ -0,0 +1,56 @@ +from dataclasses import dataclass +from datetime import datetime +from typing import Dict, Any, Optional, Sequence + +from .base_entity_request import BaseEntityRequest + + +@dataclass +class LineReportRequest(BaseEntityRequest): + """ + Request class for Line Report queries. + + Attributes + ---------- + since : Optional[datetime], optional + Filter line reports since this date and time (default is None). + until : Optional[datetime], optional + Filter line reports until this date and time (default is None). + depth : int, optional + The depth of the query (default is 1). + forbidden_uris : Optional[Sequence[str]], optional + List of URIs to forbid (default is None). + disable_geojson : bool, optional + Whether to disable GeoJSON output (default is False). + """ + + since: Optional[datetime] = None + until: Optional[datetime] = None + depth: int = 1 + forbidden_uris: Optional[Sequence[str]] = None + disable_geojson: bool = False + + def to_filters(self) -> Dict[str, Any]: + """ + Converts the LineReportRequest instance into a dictionary of filters. + + Returns + ------- + Dict[str, Any] + A dictionary representation of the request filters. + """ + filters: Dict[str, Any] = { + "count": self.count, + "depth": self.depth, + } + + if self.disable_geojson: + filters["disable_geojson"] = self.disable_geojson + if self.forbidden_uris: + filters["forbidden_uris[]"] = self.forbidden_uris + if self.since: + filters["since"] = self.since.isoformat() + if self.until: + filters["until"] = self.until.isoformat() + + return filters diff --git a/navitia_client/entities/request/place.py b/navitia_client/entities/request/place.py new file mode 100644 index 0000000..72a4bec --- /dev/null +++ b/navitia_client/entities/request/place.py @@ -0,0 +1,52 @@ +from dataclasses import dataclass +from typing import Dict, Any, Optional, Sequence, Tuple + +from .base_entity_request import BaseEntityRequest + + +@dataclass +class PlaceRequest(BaseEntityRequest): + """ + Request class for Place queries. + + Attributes + ---------- + query : str + The search query string for places. + type : Sequence[str], optional + The types of places to search for (default is ["stop_area", "address", "poi", "administrative_region"]). + disable_geojson : bool, optional + Whether to disable GeoJSON in the response (default is False). + depth : int, optional + The depth of data to retrieve (default is 1). + from_lon_lat : Optional[Tuple[float, float]], optional + Longitude and latitude coordinates to filter from (default is None). + """ + + query: str + type: Sequence[str] = ("stop_area", "address", "poi", "administrative_region") + disable_geojson: bool = False + depth: int = 1 + from_lon_lat: Optional[Tuple[float, float]] = None + + def to_filters(self) -> Dict[str, Any]: + """ + Converts the PlaceRequest instance into a dictionary of filters. + + Returns + ------- + Dict[str, Any] + A dictionary representation of the request filters. + """ + filters = { + "q": self.query, + "type": self.type, + "depth": self.depth, + } + + if self.disable_geojson: + filters["disable_geojson"] = self.disable_geojson + if self.from_lon_lat: + filters["filter"] = f"{self.from_lon_lat[0]};{self.from_lon_lat[1]}" + + return filters diff --git a/navitia_client/entities/request/places_nearby.py b/navitia_client/entities/request/places_nearby.py new file mode 100644 index 0000000..e816bdb --- /dev/null +++ b/navitia_client/entities/request/places_nearby.py @@ -0,0 +1,68 @@ +from dataclasses import dataclass +from typing import Dict, Any, Optional, Sequence + +from .base_entity_request import BaseEntityRequest + + +@dataclass +class PlacesNearbyRequest(BaseEntityRequest): + """ + Request class for Places Nearby queries. + + Attributes + ---------- + distance : int, optional + The distance for nearby search (default is 500). + type : Sequence[str], optional + The types of places to include in the search (default is ["stop_area", "stop_point", "poi"]). + admin_uri : Optional[Sequence[str]], optional + The administrative URIs to filter by (default is None). + filter : Optional[str], optional + Additional filtering criteria (default is None). + disable_geojson : bool, optional + Whether to disable GeoJSON format in the response (default is False). + disable_disruption : bool, optional + Whether to disable disruption information (default is False). + depth : int, optional + The depth of data to retrieve (default is 1). + add_poi_infos : Sequence[str], optional + Additional POI information to include (default is ["bss_stands", "car_park"]). + """ + + 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 + add_poi_infos: Sequence[str] = ("bss_stands", "car_park") + + def to_filters(self) -> Dict[str, Any]: + """ + Converts the PlacesNearbyRequest instance into a dictionary of filters. + + Returns + ------- + Dict[str, Any] + A dictionary representation of the request filters. + """ + filters = { + "start_page": self.start_page, + "count": self.count, + "depth": self.depth, + "type[]": self.type, + "distance": self.distance, + "add_poi_infos[]": self.add_poi_infos, + } + + if self.disable_geojson: + filters["disable_geojson"] = self.disable_geojson + if self.disable_disruption: + filters["disable_disruption"] = self.disable_disruption + if self.admin_uri: + filters["admin_uris[]"] = self.admin_uri + if self.filter: + filters["filter"] = self.filter + + return filters diff --git a/navitia_client/entities/request/public_transport_object.py b/navitia_client/entities/request/public_transport_object.py new file mode 100644 index 0000000..8d676d4 --- /dev/null +++ b/navitia_client/entities/request/public_transport_object.py @@ -0,0 +1,52 @@ +from dataclasses import dataclass +from typing import Dict, Any, Optional, Sequence + +from .base_entity_request import BaseEntityRequest + + +@dataclass +class PublicTransportObjectRequest(BaseEntityRequest): + """ + Request class for Public Transport Object queries. + + Attributes + ---------- + query : str + The search query string for public transport objects. + type : Sequence[str], optional + The types of objects to search for (default is ["network", "commercial_mode", "line", "route", "stop_area"]). + disable_disruption : bool, optional + Whether to disable disruption information (default is False). + depth : int, optional + The depth of data to retrieve (default is 1). + post_query_filter : Optional[str], optional + Additional filtering criteria (default is None). + """ + + query: str + type: Sequence[str] = ("network", "commercial_mode", "line", "route", "stop_area") + disable_disruption: bool = False + depth: int = 1 + post_query_filter: Optional[str] = None + + def to_filters(self) -> Dict[str, Any]: + """ + Converts the PublicTransportObjectRequest instance into a dictionary of filters. + + Returns + ------- + Dict[str, Any] + A dictionary representation of the request filters. + """ + filters = { + "q": self.query, + "type": self.type, + "depth": self.depth, + } + + if self.disable_disruption: + filters["disable_disruption"] = self.disable_disruption + if self.post_query_filter: + filters["filter"] = self.post_query_filter + + return filters diff --git a/navitia_client/entities/request/route_schedule.py b/navitia_client/entities/request/route_schedule.py new file mode 100644 index 0000000..3f9053c --- /dev/null +++ b/navitia_client/entities/request/route_schedule.py @@ -0,0 +1,67 @@ +from dataclasses import dataclass, field +from datetime import datetime +from typing import Dict, Any, Optional, Sequence + +from .base_entity_request import BaseEntityRequest + + +@dataclass +class RouteScheduleRequest(BaseEntityRequest): + """ + Request class for RouteSchedule entities. + + Attributes + ---------- + from_datetime : datetime, optional + The starting datetime for fetching route schedules (default is current datetime). + duration : int, optional + The duration in seconds for which to fetch route schedules (default is 86400 seconds). + depth : int, optional + The depth of the search (default is 1). + items_per_schedule : int, optional + The number of items per schedule (default is 1). + forbidden_uris : Optional[Sequence[str]], optional + A list of URIs to exclude from the search (default is None). + data_freshness : str, optional + The freshness of the data to fetch, either "realtime" or "base_schedule" (default is "base_schedule"). + disable_geojson : bool, optional + Whether to disable geoJSON in the response (default is False). + direction_type : str, optional + The direction type of the route schedules to fetch, e.g., "all", "forward", "backward" (default is "all"). + """ + + from_datetime: datetime = field(default_factory=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" + + def to_filters(self) -> Dict[str, Any]: + """ + Converts the RouteScheduleRequest instance into a dictionary of filters. + + Returns + ------- + Dict[str, Any] + A dictionary representation of the request filters. + """ + filters = { + "from_datetime": self.from_datetime, + "duration": self.duration, + "depth": self.depth, + "count": self.count, + "start_page": self.start_page, + "items_per_schedule": self.items_per_schedule, + "data_freshness": self.data_freshness, + "direction_type": self.direction_type, + } + + if self.disable_geojson: + filters["disable_geojson"] = self.disable_geojson + if self.forbidden_uris: + filters["forbidden_uris[]"] = self.forbidden_uris + + return filters diff --git a/navitia_client/entities/request/stop_schedule.py b/navitia_client/entities/request/stop_schedule.py new file mode 100644 index 0000000..a38533f --- /dev/null +++ b/navitia_client/entities/request/stop_schedule.py @@ -0,0 +1,67 @@ +from dataclasses import dataclass, field +from datetime import datetime +from typing import Dict, Any, Optional, Sequence + +from .base_entity_request import BaseEntityRequest + + +@dataclass +class StopScheduleRequest(BaseEntityRequest): + """ + Request class for StopSchedule entities. + + Attributes + ---------- + from_datetime : datetime, optional + The starting datetime for fetching stop schedules (default is current datetime). + duration : int, optional + The duration in seconds for which to fetch stop schedules (default is 86400 seconds). + depth : int, optional + The depth of the search (default is 1). + items_per_schedule : int, optional + The number of items per schedule (default is 1). + forbidden_uris : Optional[Sequence[str]], optional + A list of URIs to exclude from the search (default is None). + data_freshness : str, optional + The freshness of the data to fetch, either "realtime" or "base_schedule" (default is "realtime"). + disable_geojson : bool, optional + Whether to disable geoJSON in the response (default is False). + direction_type : str, optional + The direction type of the stop schedules to fetch, e.g., "all", "forward", "backward" (default is "all"). + """ + + from_datetime: datetime = field(default_factory=datetime.now) + duration: int = 86400 + depth: int = 1 + items_per_schedule: int = 1 + forbidden_uris: Optional[Sequence[str]] = None + data_freshness: str = "realtime" + disable_geojson: bool = False + direction_type: str = "all" + + def to_filters(self) -> Dict[str, Any]: + """ + Converts the StopScheduleRequest instance into a dictionary of filters. + + Returns + ------- + Dict[str, Any] + A dictionary representation of the request filters. + """ + filters = { + "from_datetime": self.from_datetime, + "duration": self.duration, + "depth": self.depth, + "count": self.count, + "start_page": self.start_page, + "items_per_schedule": self.items_per_schedule, + "data_freshness": self.data_freshness, + "direction_type": self.direction_type, + } + + if self.disable_geojson: + filters["disable_geojson"] = self.disable_geojson + if self.forbidden_uris: + filters["forbidden_uris[]"] = self.forbidden_uris + + return filters diff --git a/navitia_client/entities/request/terminus_schedule.py b/navitia_client/entities/request/terminus_schedule.py new file mode 100644 index 0000000..af3a57b --- /dev/null +++ b/navitia_client/entities/request/terminus_schedule.py @@ -0,0 +1,67 @@ +from dataclasses import dataclass, field +from datetime import datetime +from typing import Dict, Any, Optional, Sequence + +from .base_entity_request import BaseEntityRequest + + +@dataclass +class TerminusScheduleRequest(BaseEntityRequest): + """ + Request class for TerminusSchedule entities. + + Attributes + ---------- + from_datetime : datetime, optional + The starting datetime for fetching terminus schedules (default is current datetime). + duration : int, optional + The duration in seconds for which to fetch terminus schedules (default is 86400 seconds). + depth : int, optional + The depth of the search (default is 1). + items_per_schedule : int, optional + The number of items per schedule (default is 1). + forbidden_uris : Optional[Sequence[str]], optional + A list of URIs to exclude from the search (default is None). + data_freshness : str, optional + The freshness of the data to fetch, either "realtime" or "base_schedule" (default is "realtime"). + disable_geojson : bool, optional + Whether to disable geoJSON in the response (default is False). + direction_type : str, optional + The direction type of the terminus schedules to fetch, e.g., "all", "forward", "backward" (default is "all"). + """ + + from_datetime: datetime = field(default_factory=datetime.now) + duration: int = 86400 + depth: int = 1 + items_per_schedule: int = 1 + forbidden_uris: Optional[Sequence[str]] = None + data_freshness: str = "realtime" + disable_geojson: bool = False + direction_type: str = "all" + + def to_filters(self) -> Dict[str, Any]: + """ + Converts the TerminusScheduleRequest instance into a dictionary of filters. + + Returns + ------- + Dict[str, Any] + A dictionary representation of the request filters. + """ + filters = { + "from_datetime": self.from_datetime, + "duration": self.duration, + "depth": self.depth, + "count": self.count, + "start_page": self.start_page, + "items_per_schedule": self.items_per_schedule, + "data_freshness": self.data_freshness, + "direction_type": self.direction_type, + } + + if self.disable_geojson: + filters["disable_geojson"] = self.disable_geojson + if self.forbidden_uris: + filters["forbidden_uris[]"] = self.forbidden_uris + + return filters diff --git a/navitia_client/entities/request/traffic_report.py b/navitia_client/entities/request/traffic_report.py new file mode 100644 index 0000000..9e69df8 --- /dev/null +++ b/navitia_client/entities/request/traffic_report.py @@ -0,0 +1,56 @@ +from dataclasses import dataclass +from datetime import datetime +from typing import Dict, Any, Optional, Sequence + +from .base_entity_request import BaseEntityRequest + + +@dataclass +class TrafficReportRequest(BaseEntityRequest): + """ + Request class for Traffic Report queries. + + Attributes + ---------- + since : Optional[datetime], optional + The start datetime for the reports (default is None). + until : Optional[datetime], optional + The end datetime for the reports (default is None). + depth : int, optional + The depth of data to retrieve (default is 1). + forbidden_uris : Optional[Sequence[str]], optional + Forbidden URIs (default is None). + disable_geojson : bool, optional + Whether to disable GeoJSON in the response (default is False). + """ + + since: Optional[datetime] = None + until: Optional[datetime] = None + depth: int = 1 + forbidden_uris: Optional[Sequence[str]] = None + disable_geojson: bool = False + + def to_filters(self) -> Dict[str, Any]: + """ + Converts the TrafficReportRequest instance into a dictionary of filters. + + Returns + ------- + Dict[str, Any] + A dictionary representation of the request filters. + """ + filters: Dict[str, Any] = { + "count": self.count, + "depth": self.depth, + } + + if self.disable_geojson: + filters["disable_geojson"] = self.disable_geojson + if self.forbidden_uris: + filters["forbidden_uris[]"] = self.forbidden_uris + if self.since: + filters["since"] = self.since.isoformat() + if self.until: + filters["until"] = self.until.isoformat() + + return filters diff --git a/tests/client/apis/test_arrival_apis.py b/tests/client/apis/test_arrival_apis.py index 22311c4..c72b28a 100644 --- a/tests/client/apis/test_arrival_apis.py +++ b/tests/client/apis/test_arrival_apis.py @@ -6,6 +6,7 @@ from navitia_client.client.apis.arrival_apis import ( ArrivalApiClient, ) +from navitia_client.entities.request.arrival import ArrivalRequest from navitia_client.entities.response.arrival import Arrival @@ -28,8 +29,9 @@ def test_list_objects_by_id_and_path( mock_get_navitia_api.return_value = mock_response # When + request = ArrivalRequest() arrivals, _ = arrival_apis.list_arrivals_by_region_id_and_path( - region_id="bar", resource_path="foo:bar:fuzz" + region_id="bar", resource_path="foo:bar:fuzz", request=request ) # Then @@ -49,8 +51,9 @@ def test_list_objects_by_coordinates( mock_get_navitia_api.return_value = mock_response # When + request = ArrivalRequest() arrivals, _ = arrival_apis.list_arrivals_by_coordinates( - region_lon=1.1, region_lat=2.2, lat=1.3, lon=2.3 + region_lon=1.1, region_lat=2.2, lat=1.3, lon=2.3, request=request ) # Then diff --git a/tests/client/apis/test_contributors_apis.py b/tests/client/apis/test_contributors_apis.py index 250494d..fbc21f3 100644 --- a/tests/client/apis/test_contributors_apis.py +++ b/tests/client/apis/test_contributors_apis.py @@ -4,6 +4,7 @@ import pytest from navitia_client.client.apis.contributors_apis import ContributorsApiClient +from navitia_client.entities.request.contributor import ContributorRequest from navitia_client.entities.response.contributor import Contributor from navitia_client.entities.response import Pagination @@ -25,9 +26,12 @@ def test_list_contributors( mock_response.json.return_value = json.load(file) mock_get_navitia_api.return_value = mock_response + request = ContributorRequest() # When - contributors, pagination = contributors_apis.list_contributors(region_id="bar") + contributors, pagination = contributors_apis.list_contributors( + region_id="bar", request=request + ) # Then assert len(contributors) == 1 @@ -48,10 +52,11 @@ def test_get_region_by_id( with open("tests/test_data/contributors.json", encoding="utf-8") as file: mock_response.json.return_value = json.load(file) mock_get_navitia_api.return_value = mock_response + request = ContributorRequest() # When contributors, _ = contributors_apis.get_contributor_on_dataset( - region_id="bar", dataset_id="foo:xxx" + region_id="bar", dataset_id="foo:xxx", request=request ) # Then diff --git a/tests/client/apis/test_coverage_apis.py b/tests/client/apis/test_coverage_apis.py index d28da60..72ac27c 100644 --- a/tests/client/apis/test_coverage_apis.py +++ b/tests/client/apis/test_coverage_apis.py @@ -5,6 +5,7 @@ import pytest from navitia_client.client.apis.coverage_apis import CoverageApiClient +from navitia_client.entities.request.coverage import CoverageRequest from navitia_client.entities.response.administrative_region import Region from navitia_client.entities.response import Pagination @@ -26,9 +27,10 @@ def test_list_covered_areas( mock_response.json.return_value = json.load(file) mock_get_navitia_api.return_value = mock_response + request = CoverageRequest() # When - regions, pagination = coverage_apis.list_covered_areas() + regions, pagination = coverage_apis.list_covered_areas(request=request) # Then assert len(regions) == 1 @@ -71,9 +73,10 @@ def test_get_region_by_id( }, } mock_get_navitia_api.return_value = mock_response + request = CoverageRequest() # When - regions, pagination = coverage_apis.get_coverage_by_region_id("12") + regions, pagination = coverage_apis.get_coverage_by_region_id("12", request=request) # Then assert len(regions) == 1 @@ -108,10 +111,13 @@ def test_get_region_by_coordinates( }, } mock_get_navitia_api.return_value = mock_response + request = CoverageRequest() # When regions, pagination = ( - coverage_apis.get_coverage_by_region_coordinates_and_coordinates(12.5, 13.2) + coverage_apis.get_coverage_by_region_coordinates_and_coordinates( + 12.5, 13.2, request=request + ) ) # Then diff --git a/tests/client/apis/test_datasets_apis.py b/tests/client/apis/test_datasets_apis.py index 1ecc23e..3b59d25 100644 --- a/tests/client/apis/test_datasets_apis.py +++ b/tests/client/apis/test_datasets_apis.py @@ -5,6 +5,7 @@ import pytest from navitia_client.client.apis.datasets_apis import DatasetsApiClient +from navitia_client.entities.request.dataset import DatasetRequest from navitia_client.entities.response.dataset import Dataset @@ -25,9 +26,10 @@ def test_list_covered_areas( mock_response.json.return_value = json.load(file) mock_get_navitia_api.return_value = mock_response + request = DatasetRequest() # When - datasets, _ = datasets_apis.list_datasets(region_id="bar") + datasets, _ = datasets_apis.list_datasets(region_id="bar", request=request) # Then assert len(datasets) == 1 @@ -55,10 +57,11 @@ def test_get_region_by_id( mock_response.json.return_value = json.load(file) mock_get_navitia_api.return_value = mock_response + request = DatasetRequest() # When datasets, _ = datasets_apis.get_dataset_by_id( - region_id="bar", dataset_id="foo:foo-piv" + region_id="bar", dataset_id="foo:foo-piv", request=request ) # Then @@ -78,10 +81,11 @@ def test_get_region_by_id_missing_contributor( mock_response.json.return_value = json.load(file) mock_get_navitia_api.return_value = mock_response + request = DatasetRequest() # When datasets, _ = datasets_apis.get_dataset_by_id( - region_id="bar", dataset_id="foo:foo-piv" + region_id="bar", dataset_id="foo:foo-piv", request=request ) # Then diff --git a/tests/client/apis/test_departure_apis.py b/tests/client/apis/test_departure_apis.py index 47c3304..7bf91da 100644 --- a/tests/client/apis/test_departure_apis.py +++ b/tests/client/apis/test_departure_apis.py @@ -6,6 +6,7 @@ from navitia_client.client.apis.departure_apis import ( DepartureApiClient, ) +from navitia_client.entities.request.departure import DepartureRequest from navitia_client.entities.response.departure import Departure @@ -26,10 +27,11 @@ def test_list_objects_by_region_id_and_path( mock_response.json.return_value = json.load(file) mock_get_navitia_api.return_value = mock_response + request = DepartureRequest() # When departures, _ = departure_apis.list_departures_by_region_id_and_path( - region_id="bar", resource_path="foo:bar:fuzz" + region_id="bar", resource_path="foo:bar:fuzz", request=request ) # Then @@ -47,10 +49,11 @@ def test_list_objects_by_coordinates( mock_response.json.return_value = json.load(file) mock_get_navitia_api.return_value = mock_response + request = DepartureRequest() # When departures, _ = departure_apis.list_departures_by_coordinates( - region_lon=1.1, region_lat=1.2, lon=2.1, lat=2.2 + region_lon=1.1, region_lat=1.2, lon=2.1, lat=2.2, request=request ) # Then diff --git a/tests/client/apis/test_isochrones_apis.py b/tests/client/apis/test_isochrones_apis.py index 19755cf..0c01bb9 100644 --- a/tests/client/apis/test_isochrones_apis.py +++ b/tests/client/apis/test_isochrones_apis.py @@ -4,6 +4,7 @@ import pytest from navitia_client.client.apis.isochrone_apis import IsochronesApiClient +from navitia_client.entities.request.isochrone import IsochroneRequest from navitia_client.entities.response.isochrones import Isochrone @@ -26,8 +27,9 @@ def test_list_covered_areas_with_region_id( mock_get_navitia_api.return_value = mock_response # When + request = IsochroneRequest(from_="foo") isocrhones = isochrones_apis.list_isochrones_with_region_id( - region_id="bar", from_="foo" + region_id="bar", request=request ) # Then @@ -47,7 +49,8 @@ def test_list_covered_areas( mock_get_navitia_api.return_value = mock_response # When - isocrhones = isochrones_apis.list_isochrones(from_="foo") + request = IsochroneRequest(from_="foo") + isocrhones = isochrones_apis.list_isochrones(request=request) # Then assert len(isocrhones) == 1 diff --git a/tests/client/apis/test_places_apis.py b/tests/client/apis/test_places_apis.py index f0c499c..2be2e56 100644 --- a/tests/client/apis/test_places_apis.py +++ b/tests/client/apis/test_places_apis.py @@ -4,6 +4,7 @@ import pytest from navitia_client.client.apis.place_apis import PlacesApiClient +from navitia_client.entities.request.place import PlaceRequest from navitia_client.entities.response.place import Place @@ -25,7 +26,8 @@ def test_list_objects( mock_get_navitia_api.return_value = mock_response # When - places = places_apis.list_places(region_id="bar", query="DEFENSE") + request = PlaceRequest(query="DEFENSE") + places = places_apis.list_places(region_id="bar", request=request) # Then assert len(places) == 1 diff --git a/tests/client/apis/test_public_transport_objects_apis.py b/tests/client/apis/test_public_transport_objects_apis.py index c090b29..f253c31 100644 --- a/tests/client/apis/test_public_transport_objects_apis.py +++ b/tests/client/apis/test_public_transport_objects_apis.py @@ -6,6 +6,9 @@ from navitia_client.client.apis.public_transport_objects_apis import ( PublicTransportObjectsApiClient, ) +from navitia_client.entities.request.public_transport_object import ( + PublicTransportObjectRequest, +) from navitia_client.entities.response.pt_object import PtObject @@ -30,8 +33,9 @@ def test_list_objects( mock_get_navitia_api.return_value = mock_response # When + request = PublicTransportObjectRequest(query="REMY") pt_objects = pt_objects_apis.list_public_transport_objects( - region_id="bar", query="REMY" + region_id="bar", request=request ) # Then diff --git a/tests/client/apis/test_route_schedules_apis.py b/tests/client/apis/test_route_schedules_apis.py index c00b65e..13656ee 100644 --- a/tests/client/apis/test_route_schedules_apis.py +++ b/tests/client/apis/test_route_schedules_apis.py @@ -6,6 +6,7 @@ from navitia_client.client.apis.route_schedules_apis import ( RouteSchedulesApiClient, ) +from navitia_client.entities.request.route_schedule import RouteScheduleRequest from navitia_client.entities.response.route_schedule import RouteSchedule @@ -26,10 +27,11 @@ def test_list_objects_by_region_id_and_path( mock_response.json.return_value = json.load(file) mock_get_navitia_api.return_value = mock_response + request = RouteScheduleRequest() # When route_schedules = route_schedules_apis.list_route_schedules_by_region_id_and_path( - region_id="bar", resource_path="foo:bar:fuzz" + region_id="bar", resource_path="foo:bar:fuzz", request=request ) # Then @@ -47,10 +49,11 @@ def test_list_objects_by_coordinates( mock_response.json.return_value = json.load(file) mock_get_navitia_api.return_value = mock_response + request = RouteScheduleRequest() # When route_schedules = route_schedules_apis.list_route_schedules_by_coordinates( - region_lon=1.1, region_lat=1.2, lon=2.1, lat=2.2 + region_lon=1.1, region_lat=1.2, lon=2.1, lat=2.2, request=request ) # Then diff --git a/tests/client/apis/test_stop_schedules_apis.py b/tests/client/apis/test_stop_schedules_apis.py index 99904a5..90e4aa1 100644 --- a/tests/client/apis/test_stop_schedules_apis.py +++ b/tests/client/apis/test_stop_schedules_apis.py @@ -6,6 +6,7 @@ from navitia_client.client.apis.stop_schedules_apis import ( StopSchedulesApiClient, ) +from navitia_client.entities.request.stop_schedule import StopScheduleRequest from navitia_client.entities.response.stop_schedule import StopSchedule @@ -26,10 +27,11 @@ def test_list_objects_by_region_id_and_path( mock_response.json.return_value = json.load(file) mock_get_navitia_api.return_value = mock_response + request = StopScheduleRequest() # When stop_schedules, _ = stop_schedules_apis.list_stop_schedules_by_region_id_and_path( - region_id="bar", resource_path="foo:bar:fuzz" + region_id="bar", resource_path="foo:bar:fuzz", request=request ) # Then @@ -47,10 +49,11 @@ def test_list_objects_by_coordinates( mock_response.json.return_value = json.load(file) mock_get_navitia_api.return_value = mock_response + request = StopScheduleRequest() # When stop_schedules, _ = stop_schedules_apis.list_stop_schedules_by_coordinates( - region_lon=1.1, region_lat=1.2, lon=2.1, lat=2.2 + region_lon=1.1, region_lat=1.2, lon=2.1, lat=2.2, request=request ) # Then diff --git a/tests/client/apis/test_terminus_schedules_apis.py b/tests/client/apis/test_terminus_schedules_apis.py index 188b0fe..dd48637 100644 --- a/tests/client/apis/test_terminus_schedules_apis.py +++ b/tests/client/apis/test_terminus_schedules_apis.py @@ -6,6 +6,7 @@ from navitia_client.client.apis.terminus_schedules_apis import ( TerminusSchedulesApiClient, ) +from navitia_client.entities.request.terminus_schedule import TerminusScheduleRequest from navitia_client.entities.response.stop_schedule import TerminusSchedule @@ -26,11 +27,12 @@ def test_list_objects_by_region_id_and_path( mock_response.json.return_value = json.load(file) mock_get_navitia_api.return_value = mock_response + request = TerminusScheduleRequest() # When terminus_schedules, _ = ( terminus_schedules_apis.list_terminus_schedules_by_region_id_and_path( - region_id="bar", resource_path="foo:bar:fuzz" + region_id="bar", resource_path="foo:bar:fuzz", request=request ) ) @@ -49,11 +51,12 @@ def test_list_objects_by_coordinates( mock_response.json.return_value = json.load(file) mock_get_navitia_api.return_value = mock_response + request = TerminusScheduleRequest() # When terminus_schedules, _ = ( terminus_schedules_apis.list_terminus_schedules_by_coordinates( - region_lon=1.1, region_lat=1.2, lon=2.1, lat=2.2 + region_lon=1.1, region_lat=1.2, lon=2.1, lat=2.2, request=request ) )