Moved parcelsapp client

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(
+    parcelsapp_client: ParcelsAppClient = ParcelsAppClient(
-        secrets.PARCELS_API_KEY,
+        requests.Session(),
+        parcels_api_key=secrets.PARCELS_API_KEY,
     )
 
     http.initialize_server(parcelsapp_client)

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 = """
 <doctype HTML>
@@ -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)

package_tracking/parcelsapp.py

@@ -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']
-                ],
-            )

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