Add 'train' methods
This commit is contained in:
parent
af385b779d
commit
d4bec29e49
@ -16,10 +16,12 @@
|
||||
|
||||
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
|
||||
|
||||
logger = logging.getLogger('OwlBoardClient')
|
||||
logger.setLevel(logging.INFO)
|
||||
@ -67,12 +69,70 @@ class OwlBoardClient:
|
||||
print(response.text)
|
||||
|
||||
### Train Methods ###
|
||||
def get_trains_by_headcode(self, headcode: str, date: datetime):
|
||||
if not isinstance(headcode, str):
|
||||
raise TypeError("headcode must be a string")
|
||||
if not isinstance(date, datetime):
|
||||
raise TypeError("date must be a datetime object")
|
||||
|
||||
# Generate URL
|
||||
try:
|
||||
url_path = url_multijoin(
|
||||
self.base_url,
|
||||
ENDPOINTS['TIMETABLE_TRAINS'],
|
||||
format_url_datetime(date),
|
||||
'headcode',
|
||||
headcode,
|
||||
)
|
||||
logger.debug(f"Generated URL: {url_path}")
|
||||
except Exception as e:
|
||||
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
|
||||
|
||||
def get_trains_by_trainUid(self, train_uid: str, date: datetime):
|
||||
if not isinstance(train_uid, str):
|
||||
raise TypeError("train_uid must be a string")
|
||||
if not isinstance(date, datetime):
|
||||
raise TypeError("date must be a datetime object")
|
||||
|
||||
# Generate URL
|
||||
try:
|
||||
url_path = url_multijoin(
|
||||
self.base_url,
|
||||
ENDPOINTS['TIMETABLE_TRAINS'],
|
||||
format_url_datetime(date),
|
||||
'byTrainUid',
|
||||
train_uid,
|
||||
)
|
||||
logger.debug(f"Generated URL: {url_path}")
|
||||
except Exception as e:
|
||||
logger.error(f"Error generating URL: {e}")
|
||||
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
|
||||
|
||||
if __name__ == "__main__":
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
|
||||
try:
|
||||
client = OwlBoardClient(base_url='https://owlboard.info', api_key="x")
|
||||
client.get_pis_by_start_end_crs('bri', 'wsb')
|
||||
client.get_trains_by_headcode("1A99", datetime.now())
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to create client: {e}")
|
@ -4,4 +4,5 @@ ENDPOINTS = {
|
||||
'TEST': '/api/v1/auth/test/',
|
||||
'PIS_BY_CODE': '/api/v2/pis/byCode/',
|
||||
'PIS_BY_START_END_CRS': '/api/v2/pis/byStartEnd/',
|
||||
'TIMETABLE_TRAINS': '/api/v2/timetable/train'
|
||||
}
|
@ -13,3 +13,21 @@
|
||||
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
from urllib.parse import urljoin
|
||||
from datetime import datetime
|
||||
|
||||
def url_multijoin(*parts: str) -> str:
|
||||
if not parts:
|
||||
return ""
|
||||
|
||||
base = parts[0]
|
||||
for part in parts[1:]:
|
||||
base = urljoin(base.rstrip('/') + '/', part.lstrip('/'))
|
||||
|
||||
return base
|
||||
|
||||
def format_url_datetime(input: datetime) -> str:
|
||||
if not isinstance(input, datetime):
|
||||
raise TypeError("Expected a datetime object")
|
||||
return input.strftime("%Y-%m-%d")
|
Loading…
Reference in New Issue
Block a user