50 lines
1.6 KiB
Python
50 lines
1.6 KiB
Python
|
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'])
|
||
|
|
|