OwlBoard/pyOwlBoard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Activity

Refactor OwlBoardClient

Browse Source
This commit is contained in:
Fred Boniface 2024-10-29 13:34:07 +00:00
parent 710e61bb17
commit 767e258687
1 changed files with 28 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
pyOwlBoard/client.py
View File

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