From 66e3d5e8a70fb4e585641f38d4865433626f9537 Mon Sep 17 00:00:00 2001 From: Jon Michael Aanes Date: Sun, 4 May 2025 23:46:52 +0200 Subject: [PATCH] Moved parcelsapp client --- package_tracking/__main__.py | 9 ++- package_tracking/http.py | 12 ++-- package_tracking/parcelsapp.py | 101 --------------------------------- test/test_init.py | 1 - 4 files changed, 13 insertions(+), 110 deletions(-) delete mode 100644 package_tracking/parcelsapp.py diff --git a/package_tracking/__main__.py b/package_tracking/__main__.py index 0ab8faf..524500a 100644 --- a/package_tracking/__main__.py +++ b/package_tracking/__main__.py @@ -1,6 +1,8 @@ import logging -from . import http, parcelsapp, secrets +import requests +from . import http, secrets +from clients.parcelsapp import ParcelsAppClient logger = logging.getLogger(__name__) @@ -9,8 +11,9 @@ def main(): logging.basicConfig() logger.setLevel('INFO') - parcelsapp_client: parcelsapp.ParcelsAppClient = parcelsapp.ParcelsAppClient( - secrets.PARCELS_API_KEY, + parcelsapp_client: ParcelsAppClient = ParcelsAppClient( + requests.Session(), + parcels_api_key=secrets.PARCELS_API_KEY, ) http.initialize_server(parcelsapp_client) diff --git a/package_tracking/http.py b/package_tracking/http.py index 1dfb357..5d1950b 100644 --- a/package_tracking/http.py +++ b/package_tracking/http.py @@ -1,9 +1,11 @@ import bottle import datetime -from . import database, parcelsapp +from . import database +from clients.parcelsapp import ParcelsAppClient +from clients.common import ApiError -PARCELSAPP_CLIENT: parcelsapp.ParcelsAppClient | None = None +PARCELSAPP_CLIENT: ParcelsAppClient | None = None TEMPLATE = """ @@ -149,7 +151,7 @@ def get_packages_from_parcels(): try: tracking_results = list(PARCELSAPP_CLIENT.get_tracking_status(tracking_numbers)) error_message = None - except parcelsapp.ParcelsApiError as ex: + except ApiError as ex: tracking_results = [] error_message = str(ex) @@ -160,7 +162,7 @@ def get_packages_from_parcels(): return [(e, tracking_results_by_id.get(e.number)) for e in tracking_entries] -def render_tracking(tracking_results_with_name, error_message: str | None, with_form: bool): +def render_tracking(error_message: str | None = None, with_form: bool = False): tracking_results_with_name = get_packages_from_parcels() tracking_results_with_name.sort(key=lambda x: x[1].latest_state().date if x[1] else TODAY, reverse=True) return bottle.template(TEMPLATE, error_message=error_message, tracking_results=tracking_results_with_name, with_form=with_form) @@ -185,7 +187,7 @@ def add_tracking_number(): return bottle.redirect('/') -def initialize_server(parcelsapp_client: parcelsapp.ParcelsAppClient): +def initialize_server(parcelsapp_client: ParcelsAppClient): global PARCELSAPP_CLIENT PARCELSAPP_CLIENT = parcelsapp_client bottle.run(host='0.0.0.0', port=8080, debug=False) diff --git a/package_tracking/parcelsapp.py b/package_tracking/parcelsapp.py deleted file mode 100644 index a55c8cf..0000000 --- a/package_tracking/parcelsapp.py +++ /dev/null @@ -1,101 +0,0 @@ -import dataclasses -import datetime -import logging -import time -from collections.abc import Iterator - -import requests - -logger = logging.getLogger(__name__) - -URL_TRACKING = 'https://parcelsapp.com/api/v3/shipments/tracking' - -target_country = 'Denmark' - -TRACKING_STATUS_CHECKING_INTERVAL = 1 - - -@dataclasses.dataclass(frozen=True) -class ParcelState: - date: datetime.datetime - status: str - carrier: str | None - - -@dataclasses.dataclass(frozen=True) -class ParcelInfo: - tracking_number: str - tracking_url: str - status: str - destination: str | None - origin: str | None - states: list[ParcelState] - # TODO: More fields - - def latest_state(self) -> ParcelState: - return max(self.states, key=lambda state: state.date) - -class ParcelsApiError(RuntimeError): - pass - -class ParcelsAppClient: - def __init__(self, api_key: str): - assert api_key is not None, 'Missing API Key' - self.api_key = api_key - - def _request_json(self, method: str, url: str, **kwargs) -> dict: - request_json_data = {'apiKey': self.api_key, 'language': 'en', **kwargs} - response = requests.request( - method=method, url=url, json=request_json_data, - ) - response.raise_for_status() - json_data = response.json() - if 'error' in json_data: - msg = 'Error from endpoint: {}'.format(json_data['error']) - raise ParcelsApiError(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_to_json(self, tracking_ids: list[str]) -> dict: - shipments = [ - {'trackingId': tracking_id, 'country': target_country} - for tracking_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']) - - def get_tracking_status(self, tracking_ids: list[str]) -> Iterator[ParcelInfo]: - if len(tracking_ids) == 0: - return - - for parcel_json in self._get_tracking_status_to_json(tracking_ids)['shipments']: - yield ParcelInfo( - tracking_number=parcel_json['trackingId'], - tracking_url=parcel_json['externalTracking'][0]['url'], - status=parcel_json['status'], - destination=parcel_json.get('destination'), - origin=parcel_json.get('origin'), - states=[ - ParcelState( - status=s['status'], - date=datetime.datetime.fromisoformat(s['date']), - carrier=s['carrier'], - ) - for s in parcel_json['states'] - ], - ) diff --git a/test/test_init.py b/test/test_init.py index 02fc09a..557bedf 100644 --- a/test/test_init.py +++ b/test/test_init.py @@ -3,4 +3,3 @@ def test_import_modules(): import package_tracking # noqa import package_tracking.database # noqa import package_tracking.http # noqa - import package_tracking.parcelsapp # noqa