DX-1084 - Test track by tracking number and carrier code.

shipengine_sdk/http_client/client.py

@@ -2,7 +2,7 @@
 import json
 import os
 import platform
-from typing import Dict, Optional
+from typing import Any, Dict, Optional
 
 import requests
 from requests import PreparedRequest, Request, RequestException, Response, Session
@@ -37,8 +37,8 @@ def __init__(self) -> None:
         self.session = requests.session()
 
     def send_rpc_request(
-        self, method: str, params: Optional[Dict[str, any]], retry: int, config: ShipEngineConfig
-    ) -> Dict[str, any]:
+        self, method: str, params: Optional[Dict[str, Any]], retry: int, config: ShipEngineConfig
+    ) -> Dict[str, Any]:
         """
         Send a `JSON-RPC 2.0` request via HTTP Messages to ShipEngine API. If the response
          * is successful, the result is returned. Otherwise, an error is thrown.
@@ -53,13 +53,13 @@ def send_rpc_request(
             else os.getenv("CLIENT_BASE_URI")
         )
 
-        request_headers: Dict[str, any] = {
+        request_headers: Dict[str, Any] = {
             "User-Agent": self._derive_user_agent(),
             "Content-Type": "application/json",
             "Accept": "application/json",
         }
 
-        request_body: Dict[str, any] = wrap_request(method=method, params=params)
+        request_body: Dict[str, Any] = wrap_request(method=method, params=params)
 
         req: Request = Request(
             method="POST",
@@ -80,7 +80,7 @@ def send_rpc_request(
                 error_code=ErrorCode.UNSPECIFIED.value,
             )
 
-        resp_body: Dict[str, any] = resp.json()
+        resp_body: Dict[str, Any] = resp.json()
         status_code: int = resp.status_code
 
         is_response_404(status_code=status_code, response_body=resp_body, config=config)

shipengine_sdk/jsonrpc/__init__.py

@@ -3,7 +3,7 @@
 functionality for sending HTTP requests from the ShipEngine SDK.
 """
 import time
-from typing import Dict, Optional
+from typing import Any, Dict, Optional
 
 from ..errors import RateLimitExceededError
 from ..http_client import ShipEngineClient
@@ -12,22 +12,24 @@
 
 
 def rpc_request(
-    method: str, config: ShipEngineConfig, params: Optional[Dict[str, any]] = None
-) -> Dict[str, any]:
+    method: str, config: ShipEngineConfig, params: Optional[Dict[str, Any]] = None
+) -> Dict[str, Any]:
     """
     Create and send a `JSON-RPC 2.0` request over HTTP messages.
     TODO: add param and return docs
     """
     return rpc_request_loop(method, params, config)
 
 
-def rpc_request_loop(method: str, params: dict, config: ShipEngineConfig) -> Dict[str, any]:
+def rpc_request_loop(
+    method: str, params: Optional[Dict[str, Any]], config: ShipEngineConfig
+) -> Dict[str, Any]:
     client: ShipEngineClient = ShipEngineClient()
-    api_response: Optional[Dict[str, any]] = None
+    api_response: Optional[Dict[str, Any]] = None
     retry: int = 0
     while retry <= config.retries:
         try:
-            api_response: Dict[str, any] = client.send_rpc_request(
+            api_response = client.send_rpc_request(
                 method=method, params=params, retry=retry, config=config
             )
         except Exception as err:

shipengine_sdk/jsonrpc/process_request.py

@@ -1,5 +1,5 @@
 """Functions that help with process requests and handle responses."""
-from typing import Dict, Optional
+from typing import Any, Dict, Optional
 from uuid import uuid4
 
 from ..errors import (
@@ -13,15 +13,15 @@
 from ..models import ErrorType
 
 
-def wrap_request(method: str, params: Optional[Dict[str, any]]) -> Dict[str, any]:
+def wrap_request(method: str, params: Optional[Dict[str, Any]]) -> Dict[str, Any]:
     """
     Wrap request per `JSON-RPC 2.0` spec.
 
     :param str method: The RPC Method to be sent to the RPC Server to
     invoke a specific remote procedure.
     :param params: The request data for the RPC request. This argument
     is optional and can either be a dictionary or None.
-    :type params: Optional[Dict[str, any]]
+    :type params: Optional[Dict[str, Any]]
     """
     if params is None:
         return dict(id=f"req_{str(uuid4()).replace('-', '')}", jsonrpc="2.0", method=method)
@@ -31,13 +31,13 @@ def wrap_request(method: str, params: Optional[Dict[str, any]]) -> Dict[str, any
         )
 
 
-def handle_response(response_body: Dict[str, any]) -> Dict[str, any]:
+def handle_response(response_body: Dict[str, Any]) -> Dict[str, Any]:
     """Handles the response from ShipEngine API."""
     if "result" in response_body:
         return response_body
 
-    error: Dict[str, any] = response_body["error"]
-    error_data: Dict[str, any] = error["data"]
+    error: Dict[str, Any] = response_body["error"]
+    error_data: Dict[str, Any] = error["data"]
     error_type: str = error_data["type"]
     if error_type is ErrorType.ACCOUNT_STATUS.value:
         raise AccountStatusError(

shipengine_sdk/models/__init__.py

@@ -2,4 +2,11 @@
 from .address import Address, AddressValidateResult
 from .carriers import Carrier, CarrierAccount
 from .enums import *  # noqa
-from .track_pacakge import Package, Shipment, TrackingQuery, TrackPackageResult
+from .track_pacakge import (
+    Location,
+    Package,
+    Shipment,
+    TrackingEvent,
+    TrackingQuery,
+    TrackPackageResult,
+)

shipengine_sdk/models/carriers/__init__.py

@@ -1,6 +1,6 @@
 """CarrierAccount class object and immutable carrier object."""
 import json
-from typing import Dict
+from typing import Any, Dict
 
 from ...errors import InvalidFieldValueError, ShipEngineError
 from ..enums import Carriers, does_member_value_exist, get_carrier_name_value
@@ -25,7 +25,7 @@ def to_json(self):
 class CarrierAccount:
     carrier: Carrier
 
-    def __init__(self, account_information: Dict[str, any]) -> None:
+    def __init__(self, account_information: Dict[str, Any]) -> None:
         """This class represents a given account with a Carrier provider e.g. `FedEx`, `UPS`, `USPS`."""
         self._set_carrier(account_information["carrierCode"])
         self.account_id = account_information["accountID"]

shipengine_sdk/models/enums/__init__.py

@@ -6,6 +6,7 @@
 from .error_code import ErrorCode
 from .error_source import ErrorSource
 from .error_type import ErrorType
+from .regex_patterns import RegexPatterns
 
 
 class Endpoints(Enum):

shipengine_sdk/models/enums/regex_patterns.py

@@ -0,0 +1,7 @@
+"""Enumeration of Regular Expression patterns used in the ShipEngine SDK."""
+from enum import Enum
+
+
+class RegexPatterns(Enum):
+    VALID_ISO_STRING = r"^(-?(?:[1-9][0-9]*)?[0-9]{4})-(1[0-2]|0[1-9])-(3[01]|0[1-9]|[12][0-9])T(2[0-3]|[01][0-9]):([0-5][0-9]):([0-5][0-9])(\.[0-9]+)?(Z|[+-](?:2[0-3]|[01][0-9]):[0-5][0-9])?$"  # noqa
+    VALID_ISO_STRING_NO_TZ = r"^(-?(?:[1-9][0-9]*)?[0-9]{4})-(1[0-2]|0[1-9])-(3[01]|0[1-9]|[12][0-9])T(2[0-3]|[01][0-9]):([0-5][0-9]):([0-5][0-9])(\.[0-9]+)?([+-](?:2[0-3]|[01][0-9]):[0-5][0-9])?$"  # noqa

shipengine_sdk/models/track_pacakge/__init__.py

@@ -1,10 +1,9 @@
 """Data objects to be used in the `track_package` and `track` methods."""
 import json
 from dataclasses import dataclass
-from datetime import datetime
-from typing import Dict, List, Optional
+from typing import Any, Dict, List, Optional, Union
 
-from dataclasses_json import LetterCase, dataclass_json
+from dataclasses_json import LetterCase, dataclass_json  # type: ignore
 
 from ...errors import ShipEngineError
 from ...services.get_carrier_accounts import GetCarrierAccounts
@@ -15,25 +14,25 @@
 
 class Shipment:
     config: ShipEngineConfig
-    shipment_id: Optional[str]
-    account_id: Optional[str]
-    carrier_account: Optional[CarrierAccount]
+    shipment_id: Optional[str] = None
+    account_id: Optional[str] = None
+    carrier_account: Optional[CarrierAccount] = None
     carrier: Carrier
-    estimated_delivery_date: datetime
-    actual_delivery_date: datetime
+    estimated_delivery_date: Union[IsoString, str]
+    actual_delivery_date: Union[IsoString, str]
 
     def __init__(
-        self, shipment: Dict[str, any], actual_delivery_date: datetime, config: ShipEngineConfig
+        self, shipment: Dict[str, Any], actual_delivery_date: IsoString, config: ShipEngineConfig
     ) -> None:
         """This object represents a given Shipment."""
         self.config = config
         self.shipment_id = shipment["shipmentID"] if "shipmentID" in shipment else None
         self.account_id = shipment["carrierAccountID"] if "carrierAccountID" in shipment else None
-        self.carrier_account = (
-            self._get_carrier_account(carrier=shipment["carrierCode"], account_id=self.account_id)
-            if "carrierCode" in shipment
-            else None
-        )
+
+        if self.account_id is not None:
+            self.carrier_account = self._get_carrier_account(
+                carrier=shipment["carrierCode"], account_id=self.account_id
+            )
 
         if self.carrier_account is not None:
             self.carrier = self.carrier_account.carrier
@@ -44,9 +43,7 @@ def __init__(
                 else ShipEngineError("The carrierCode field was null from api response.")
             )
 
-        self.estimated_delivery_date = IsoString(
-            iso_string=shipment["estimatedDelivery"]
-        ).to_datetime_object()
+        self.estimated_delivery_date = IsoString(iso_string=shipment["estimatedDelivery"])
         self.actual_delivery_date = actual_delivery_date
 
     def _get_carrier_account(self, carrier: str, account_id: str) -> CarrierAccount:
@@ -57,33 +54,51 @@ def _get_carrier_account(self, carrier: str, account_id: str) -> CarrierAccount:
         )
 
         for account in carrier_accounts:
-            if account_id == account["account_id"]:
+            if account_id == account.account_id:
                 target_carrier.append(account)
                 return target_carrier[0]
             else:
                 raise ShipEngineError(
-                    message=f"accountID [{account_id}] doesn't amtch any of the accounts connected to your "
-                    + "ShipEngine Account."
+                    message=f"accountID [{account_id}] doesn't match any of the accounts connected to your ShipEngine Account."  # noqa
                 )
 
-    def to_dict(self):
+    def to_dict(self) -> Dict[str, Any]:
+        if hasattr(self, "config"):
+            del self.config
+        else:
+            pass  # noqa
         return (lambda o: o.__dict__)(self)
 
-    def to_json(self):
+    def to_json(self) -> str:
+        if hasattr(self, "config"):
+            del self.config
+        else:
+            pass  # noqa
         return json.dumps(self, default=lambda o: o.__dict__, indent=2)
 
 
-@dataclass_json(letter_case=LetterCase.CAMEL)
-@dataclass
 class Package:
     """This object contains package information for a given shipment."""
 
     package_id: Optional[str]
-    weight: Optional[Dict[str, any]]
-    dimensions: Optional[Dict[str, any]]
+    weight: Optional[Dict[str, Any]]
+    dimensions: Optional[Dict[str, Any]]
     tracking_number: Optional[str]
     tracking_url: Optional[str]
 
+    def __init__(self, package: Dict[str, Any]) -> None:
+        self.package_id = package["packageID"] if "packageID" in package else None
+        self.weight = package["weight"] if "weight" in package else None
+        self.dimensions = package["dimensions"] if "dimensions" in package else None
+        self.tracking_number = package["trackingNumber"] if "trackingNumber" in package else None
+        self.tracking_url = package["trackingURL"] if "trackingURL" in package else None
+
+    def to_dict(self) -> Dict[str, Any]:
+        return (lambda o: o.__dict__)(self)
+
+    def to_json(self) -> str:
+        return json.dumps(self, default=lambda o: o.__dict__, indent=2)
+
 
 @dataclass_json(letter_case=LetterCase.CAMEL)
 @dataclass
@@ -97,31 +112,29 @@ class TrackingQuery:
 @dataclass_json(letter_case=LetterCase.CAMEL)
 @dataclass
 class Location:
-    city_locality: Optional[str]
-    state_province: Optional[str]
-    postal_code: Optional[str]
-    country_code: Optional[str]
-    latitude: Optional[float]
-    longitude: Optional[float]
+    city_locality: Optional[str] = None
+    state_province: Optional[str] = None
+    postal_code: Optional[str] = None
+    country_code: Optional[str] = None
+    latitude: Optional[float] = None
+    longitude: Optional[float] = None
 
 
 class TrackingEvent:
-    date_time: datetime
-    carrier_date_time: datetime
+    date_time: Union[IsoString, str]
+    carrier_date_time: Union[IsoString, str]
     status: str
     description: Optional[str]
     carrier_status_code: Optional[str]
     carrier_detail_code: Optional[str]
     signer: Optional[str]
     location: Optional[Location]
 
-    def __init__(self, event: Dict[str, any]) -> None:
+    def __init__(self, event: Dict[str, Any]) -> None:
         """Tracking event object."""
-        self.date_time = IsoString(iso_string=event["timestamp"]).to_datetime_object()
+        self.date_time = IsoString(iso_string=event["timestamp"])
 
-        self.carrier_date_time = IsoString(
-            iso_string=event["carrierTimestamp"]
-        ).to_datetime_object()
+        self.carrier_date_time = IsoString(iso_string=event["carrierTimestamp"])
 
         self.status = event["status"]
         self.description = event["description"] if "description" in event else None
@@ -141,10 +154,11 @@ def to_json(self):
 class TrackPackageResult:
     shipment: Optional[Shipment]
     package: Optional[Package]
-    events: Optional[List[TrackingEvent]]
+    events: Optional[List[TrackingEvent]] = list()
 
-    def __init__(self, api_response: Dict[str, any], config: ShipEngineConfig) -> None:
+    def __init__(self, api_response: Dict[str, Any], config: ShipEngineConfig) -> None:
         """This object is used as the return type for the `track_package` and `track` methods."""
+        self.events = list()
         result = api_response["result"]
         for event in result["events"]:
             self.events.append(TrackingEvent(event=event))
@@ -155,10 +169,10 @@ def __init__(self, api_response: Dict[str, any], config: ShipEngineConfig) -> No
                 actual_delivery_date=self.get_latest_event().date_time,
                 config=config,
             )
-            if "shipment" in api_response
+            if "shipment" in result
             else None
         )
-        self.package = Package.from_dict(result["package"]) if "package" in result else None
+        self.package = Package(result["package"]) if "package" in result else None
 
     def get_errors(self) -> List[TrackingEvent]:  # TODO: debug
         """Returns **only** the EXCEPTION events."""
@@ -178,7 +192,15 @@ def has_errors(self) -> bool:  # TODO: debug
             return event.status == "EXCEPTION"
 
     def to_dict(self):
+        if hasattr(self.shipment, "config"):
+            del self.shipment.config
+        else:
+            pass  # noqa
         return (lambda o: o.__dict__)(self)
 
     def to_json(self):
+        if hasattr(self.shipment, "config"):
+            del self.shipment.config
+        else:
+            pass  # noqa
         return json.dumps(self, default=lambda o: o.__dict__, indent=2)