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 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, ) 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_to_json(self, tracking_ids: list[str]) -> dict: 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']) 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'] ], )