Skip to content

feat(api): replace dict by request objects for non-pt APIs #94

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jonperron merged 1 commit into from
Nov 11, 2025
Merged

feat(api): replace dict by request objects for non-pt APIs #94

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 13 additions & 86 deletions navitia_client/client/apis/arrival_apis.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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

Expand Down Expand Up @@ -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.
Expand All @@ -96,61 +88,26 @@ 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").

Copy link

Copilot AI Nov 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Missing documentation for the request parameter. Should add parameter description similar to other updated methods. Expected format:\npython\nrequest : ArrivalRequest\n The request object containing query parameters such as from_datetime,\n duration, depth, count, start_page, forbidden_uris, data_freshness,\n disable_geojson, and direction_type.\n

Suggested change
request : ArrivalRequest
The request object containing query parameters such as from_datetime,
duration, depth, count, start_page, forbidden_uris, data_freshness,
disable_geojson, and direction_type.

Copilot uses AI. Check for mistakes.

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,
region_lon: float,
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.
Expand All @@ -165,43 +122,13 @@ 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
-------
Tuple[Sequence[Arrival], Pagination]
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())
35 changes: 12 additions & 23 deletions navitia_client/client/apis/contributors_apis.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,26 +1,17 @@
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


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
Expand All @@ -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.
Expand All @@ -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"])
Expand All @@ -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.
Expand All @@ -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"])
Expand Down
49 changes: 17 additions & 32 deletions navitia_client/client/apis/coverage_apis.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,29 +1,17 @@
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


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
Expand All @@ -47,33 +35,32 @@ 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)
pagination = Pagination.from_payload(results.json()["pagination"])
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.
Expand All @@ -82,26 +69,25 @@ 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)
pagination = Pagination.from_payload(results.json()["pagination"])
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.
Expand All @@ -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)
Expand Down
Loading
Loading