diff --git a/pyOwlBoard/client.py b/pyOwlBoard/client.py index a1221bb..3957a39 100644 --- a/pyOwlBoard/client.py +++ b/pyOwlBoard/client.py @@ -18,7 +18,6 @@ import requests import logging from datetime import datetime from typing import List, Dict, Tuple -from urllib.parse import urljoin from .contants import ENDPOINTS, VERSION from .utils import format_url_datetime, url_multijoin @@ -50,24 +49,39 @@ class OwlBoardClient: raise def verify_connection(self): - url_path = urljoin(self.base_url, ENDPOINTS['TEST']) - response = requests.get(url_path, headers = self.headers) - response.raise_for_status() + url_path = url_multijoin(self.base_url, ENDPOINTS['TEST']) + response = self._make_request('GET', url_path) logger.info("Connection verified: %s", response.status_code) + def _make_request(self, method: str, url: str, **kwargs): + try: + response = requests.request(method, url, headers=self.headers, **kwargs) + response.raise_for_status() + return response + except requests.RequestException as e: + logger.error(f"Request failed: {e}") + raise + ### PIS Code Methods ### ## Needs fixing on API Server side def get_stops_by_pis(self, code: str): - url_path = urljoin(self.base_url, ENDPOINTS['PIS_BY_CODE'] + code) - response = requests.get(url_path, headers=self.headers) + url_path = url_multijoin(self.base_url, ENDPOINTS['PIS_BY_CODE'], code) + logger.debug(f"Generated URL: {url_path}") + response = self._make_request('GET', url_path) + logger.info("Response received for get_stops_by_pis") print(response.text) def get_pis_by_start_end_crs(self, start_crs: str, end_crs: str): - url_path = urljoin(self.base_url, ENDPOINTS['PIS_BY_START_END_CRS'] + start_crs + "/" + end_crs) - response = requests.get(url_path, headers=self.headers) + url_path = url_multijoin(self.base_url, ENDPOINTS['PIS_BY_START_END_CRS'], start_crs, end_crs) + logger.debug(f"Generated URL: {url_path}") + response = self._make_request('GET', url_path) + logger.info("Response received for get_pis_by_start_end_crs") print(response.text) + def get_pis_by_tiploc_list(self, tiplocs: List[str]): + return + ### Train Methods ### def get_trains_by_headcode(self, headcode: str, date: datetime): if not isinstance(headcode, str): @@ -89,14 +103,9 @@ class OwlBoardClient: logger.error(f"Error generating URL: {e}") # Send request - try: - response = requests.get(url_path, headers=self.headers) - response.raise_for_status() - logger.info("Request completed") - print(response.text) - except requests.RequestException as e: - logger.error(f"Request failed: {e}") - raise + response = self._make_request('GET', url_path) + logger.info("Response received for get_trains_by_headcode") + print(response.text) def get_trains_by_trainUid(self, train_uid: str, date: datetime): if not isinstance(train_uid, str): @@ -119,14 +128,9 @@ class OwlBoardClient: raise # Send request - try: - response = requests.get(url_path, headers=self.headers) - response.raise_for_status() - logger.info("Request completed") - print(response.text) - except requests.RequestException as e: - logger.error(f"Request failed: {e}") - raise + response = self._make_request('GET', url_path) + logger.info("Response received for get_trains_by_trainUid") + print(response.text) if __name__ == "__main__": logging.basicConfig(level=logging.DEBUG)