From ddd9d6e21f04b934b1c5afb00f2aa3427db4517b Mon Sep 17 00:00:00 2001 From: Jon Michael Aanes Date: Wed, 22 Jan 2025 14:02:02 +0100 Subject: [PATCH] Compartilizing --- package_tracking/__main__.py | 60 ++++++++++------------------------ package_tracking/http.py | 17 ++++++++++ package_tracking/parcelsapp.py | 49 +++++++++++++++++++++++++++ 3 files changed, 83 insertions(+), 43 deletions(-) create mode 100644 package_tracking/http.py create mode 100644 package_tracking/parcelsapp.py diff --git a/package_tracking/__main__.py b/package_tracking/__main__.py index b2807c2..da3f531 100644 --- a/package_tracking/__main__.py +++ b/package_tracking/__main__.py @@ -1,58 +1,32 @@ -import requests -import time import logging import yaml + +from . import parcelsapp from . import secrets +from . import http logger = logging.getLogger(__name__) -URL_TRACKING = 'https://parcelsapp.com/api/v3/shipments/tracking' +def main_cli(parcelsapp_client: parcelsapp.ParcelsAppClient): + tracking_ids = [ + '00157128965207138207', + '00057151273127784840', + ] + shipment_statuses = parcelsapp_client.get_tracking_status(tracking_ids) + print(yaml.dump(shipment_statuses['shipments'])) -target_country = 'Denmark' - -TRACKING_STATUS_CHECKING_INTERVAL = 1 - -def request_json(method: str, url: str, **kwargs) -> dict: - response = requests.request(method=method,url=URL_TRACKING, json={'apiKey': secrets.PARCELS_API_KEY, **kwargs}) - print(response) - response.raise_for_status() - json_data = response.json() - if 'error' in json_data: - msg = 'Error from endpoint: {}'.format(json_data['error']) - raise RuntimeError(msg) - return json_data - -def check_tracking_status(uuid: str) -> dict: - """ Function to check tracking status with UUID""" - json_data = request_json('GET', URL_TRACKING, uuid=uuid) - if json_data['done']: - logger.info('Tracking complete') - return json_data - else: - logger.info('Tracking in progress...') - time.sleep(TRACKING_STATUS_CHECKING_INTERVAL) - return check_tracking_status(uuid) - -def get_tracking_status(tracking_ids: list[str]): - shipments = [{'trackingId': id, 'language': 'en', 'country': target_country} for id in tracking_ids] - - # Initiate tracking request - - json_data = request_json('POST', URL_TRACKING, shipments=shipments) - if json_data.get('done'): - return json_data - return check_tracking_status(json_data['uuid']) def main(): logging.basicConfig() logger.setLevel('INFO') - tracking_ids = [ - '00157128965207138207', - '00057151273127784840', - ] - shipment_statuses = get_tracking_status(tracking_ids) - print(yaml.dump(shipment_statuses['shipments'])) + parcelsapp_client: parcelsapp.ParcelsAppClient = parcelsapp.ParcelsAppClient(secrets.PARCELS_API_KEY) + + if True: + main_cli(parcelsapp_client) + else: + http.initialize_server() + if __name__ == '__main__': main() diff --git a/package_tracking/http.py b/package_tracking/http.py new file mode 100644 index 0000000..b67edef --- /dev/null +++ b/package_tracking/http.py @@ -0,0 +1,17 @@ +from bottle import route, run, template +from . import parcelsapp + +PARCELSAPP_CLIENT: parcelsapp.ParcelsAppClient | None = None + +TEMPLATE = ''' +''' + +@route('/') +def index(): + + return template(TEMPLATE, name=name) + +def initialize_server(parcelsapp_client: parcelsapp.ParcelsAppClient): + global PARCELSAPP_CLIENT + PARCELSAPP_CLIENT = parcelsapp_client + run(host='localhost', port=8080, debug=True) diff --git a/package_tracking/parcelsapp.py b/package_tracking/parcelsapp.py new file mode 100644 index 0000000..eea1e17 --- /dev/null +++ b/package_tracking/parcelsapp.py @@ -0,0 +1,49 @@ +import requests +import time +import logging + +logger = logging.getLogger(__name__) + +URL_TRACKING = 'https://parcelsapp.com/api/v3/shipments/tracking' + +target_country = 'Denmark' + +TRACKING_STATUS_CHECKING_INTERVAL = 1 + +class ParcelsAppClient: + + def __init__(self, api_key: str): + self.api_key = api_key + + def _request_json(self, method: str, url: str, **kwargs) -> dict: + request_json_data = {'apiKey': self.api_key, **kwargs} + response = requests.request(method=method,url=URL_TRACKING, json=request_json_data) + print(response) + response.raise_for_status() + json_data = response.json() + if 'error' in json_data: + msg = 'Error from endpoint: {}'.format(json_data['error']) + raise RuntimeError(msg) + return json_data + + def check_tracking_status(self, uuid: str) -> dict: + """ Function to check tracking status with UUID""" + json_data = self._request_json('GET', URL_TRACKING, uuid=uuid) + if json_data['done']: + logger.info('Tracking complete') + return json_data + else: + logger.info('Tracking in progress...') + time.sleep(TRACKING_STATUS_CHECKING_INTERVAL) + return self.check_tracking_status(uuid) + + def get_tracking_status(self, tracking_ids: list[str]): + shipments = [{'trackingId': id, 'language': 'en', 'country': target_country} for id in tracking_ids] + + # Initiate tracking request + + json_data = self._request_json('POST', URL_TRACKING, shipments=shipments) + if json_data.get('done'): + return json_data + return self.check_tracking_status(json_data['uuid']) +