Dx 1074 get carrier accounts

shipengine_sdk/errors/__init__.py

@@ -2,7 +2,7 @@
 import json
 from typing import Optional
 
-from ..models.enums import ErrorCode, ErrorSource, ErrorType
+from ..models.enums import ErrorCode, ErrorSource, ErrorType, does_member_value_exist
 
 
 class ShipEngineError(Exception):
@@ -27,14 +27,14 @@ def __init__(
     def _are_enums_valid(self):
         if self.source is None:
             return self
-        elif self.source not in (member.value for member in ErrorSource):
+        elif not does_member_value_exist(self.source, ErrorSource):
             raise ValueError(
                 f"Error source must be a member of ErrorSource enum - [{self.source}] provided."
             )
 
         if self.error_type is None:
             return self
-        elif self.error_type not in (member.value for member in ErrorType):
+        elif not does_member_value_exist(self.error_type, ErrorType):
             raise ValueError(
                 f"Error type must be a member of ErrorType enum - [{self.error_type}] provided."
             )

shipengine_sdk/models/__init__.py

@@ -1,3 +1,13 @@
 """ShipEngine SDK Models & Enumerations"""
 from .address import Address, AddressValidateResult
-from .enums import Country, Endpoints, ErrorCode, ErrorSource, ErrorType, RPCMethods
+from .carriers import Carrier, CarrierAccount
+from .enums import (
+    CarrierNames,
+    Carriers,
+    Country,
+    Endpoints,
+    ErrorCode,
+    ErrorSource,
+    ErrorType,
+    RPCMethods,
+)

shipengine_sdk/models/address/__init__.py

@@ -27,7 +27,7 @@ class Address:
     phone: str = ""
     company: str = ""
 
-    def __post_init__(self):
+    def __post_init__(self) -> None:
         is_street_valid(self.street)
         is_city_valid(self.city_locality)
         is_state_valid(self.state_province)
@@ -52,7 +52,7 @@ def __init__(
         info: Optional[List] = None,
         warnings: Optional[List] = None,
         errors: Optional[List] = None,
-    ):
+    ) -> None:
         self.is_valid = is_valid
         self.request_id = request_id
         self.normalized_address = normalized_address

shipengine_sdk/models/carriers/__init__.py

@@ -0,0 +1,77 @@
+"""CarrierAccount class object and immutable carrier object."""
+import json
+from typing import Dict
+
+from ...errors import InvalidFieldValueError, ShipEngineError
+from ..enums import Carriers, does_member_value_exist, get_carrier_name_value
+
+# @dataclass_json(letter_case=LetterCase.CAMEL)
+# @dataclass(frozen=True)
+# class Carrier:
+#     """This immutable class object represents a given Carrier provider e.g. `FedEx`, `UPS`, `USPS`."""
+#     name: str
+#     code: str
+#
+#     def __post_init__(self) -> None:
+#         """An immutable Carrier object. e.g. `FedEx`, `UPS`, `USPS`."""
+#         if not does_member_value_exist(self.name, CarrierNames):
+#             raise ShipEngineError(f"Carrier [{self.name}] not currently supported.")
+#
+#         if not does_member_value_exist(self.code, Carriers):
+#             raise ShipEngineError(f"Carrier [{self.code}] not currently supported.")
+#
+#
+# @dataclass_json(letter_case=LetterCase.CAMEL)
+# @dataclass(frozen=True)
+# class CarrierAccount:
+#     """This class represents a given account with a Carrier provider e.g. `FedEx`, `UPS`, `USPS`."""
+#     carrier: Union[str, Carrier]
+#     account_id: str
+#     account_number: str
+#     name: str
+#
+#     def __post_init__(self) -> None:
+#         if not does_member_value_exist()
+
+
+class Carrier:
+    def __init__(self, code: str) -> None:
+        """This class represents a given account with a Carrier provider e.g. `FedEx`, `UPS`, `USPS`."""
+        if not does_member_value_exist(code, Carriers):
+            raise ShipEngineError(f"Carrier [{code}] not currently supported.")
+        else:
+            self.name = get_carrier_name_value(code.upper())
+            self.code = code
+
+    def to_dict(self):
+        return (lambda o: o.__dict__)(self)
+
+    def to_json(self):
+        return json.dumps(self, default=lambda o: o.__dict__, indent=2)
+
+
+class CarrierAccount:
+    carrier: Carrier
+
+    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"]
+        self.account_number = account_information["accountNumber"]
+
+    def _set_carrier(self, carrier: str) -> None:
+        if does_member_value_exist(carrier, Carriers):
+            self.carrier = Carrier(code=carrier).to_dict()
+            self.name = self.carrier["name"]
+        else:
+            InvalidFieldValueError(
+                field_name="carrier",
+                reason=f"Carrier [{carrier}] is currently not supported.",
+                field_value=carrier,
+            )
+
+    def to_dict(self):
+        return (lambda o: o.__dict__)(self)
+
+    def to_json(self):
+        return json.dumps(self, default=lambda o: o.__dict__, indent=2)

shipengine_sdk/models/enums/__init__.py

@@ -1,6 +1,7 @@
 """ShipEngine SDK Enumerations"""
 from enum import Enum
 
+from .carriers import CarrierNames, Carriers
 from .country import Country
 from .error_code import ErrorCode
 from .error_source import ErrorSource
@@ -21,3 +22,19 @@ class RPCMethods(Enum):
     CREATE_TAG = "create.tag.v1"
     LIST_CARRIERS = "carrier.listAccounts.v1"
     TRACK_PACKAGE = "package.track.v1"
+
+
+def does_member_value_exist(m: str, enum_to_search) -> bool:
+    """
+    Checks if a member value exists on an Enum.
+
+    :param str m: The member value to validate.
+    :param enum_to_search: The enumeration to check the member value against.
+    """
+    return False if m not in (member.value for member in enum_to_search) else True
+
+
+def get_carrier_name_value(upper_carrier_code: str):
+    for k in CarrierNames:
+        if upper_carrier_code == k.name:
+            return k.value

shipengine_sdk/models/enums/carriers/__init__.py

@@ -0,0 +1,3 @@
+"""Carrier enumerations used throughout the ShipEngine SDK."""
+from .carrier_names import CarrierNames
+from .carriers import Carriers

shipengine_sdk/models/enums/carriers/carrier_names.py

@@ -0,0 +1,63 @@
+"""Enumeration of valid carrier names."""
+from enum import Enum
+
+
+class CarrierNames(Enum):
+    """An enumeration valid carrier names."""
+
+    FEDEX = "FedEx"
+    """FedEx - Federal Express"""
+
+    UPS = "United Parcel Service"
+    """UPS - United Parcel Service"""
+
+    USPS = "U.S. Postal Service"
+    """USPS - United State Postal Service"""
+
+    STAMPS_COM = "Stamps.com"
+    """USPS services via Stamps.com"""
+
+    DHL_EXPRESS = "DHL Express"
+    """DHL Express"""
+
+    DHL_GLOBAL_MAIL = "DHL ECommerce"
+    """DHL ECommerce"""
+
+    CANADA_POST = "Canada Post"
+    """Canada Post"""
+
+    AUSTRALIA_POST = "Australia Post"
+    """Australia Post"""
+
+    FIRSTMILE = "First Mile"
+    """First Mile"""
+
+    ASENDIA = "Asendia"
+    """Asendia"""
+
+    ONTRAC = "OnTrac"
+    """OnTrac"""
+
+    APC = "APC"
+    """APC"""
+
+    NEWGISTICS = "Newgistics"
+    """Newgistics"""
+
+    GLOBEGISTICS = "Globegistics"
+    """Globegistics"""
+
+    RR_DONNELLEY = "RR Donnelley"
+    """RR Donnell"""
+
+    IMEX = "IMEX"
+    """IMEX"""
+
+    ACCESS_WORLDWIDE = "Access Worldwide"
+    """Access Worldwide"""
+
+    PUROLATOR_CA = "Purolator Canada"
+    """Purolator Canada"""
+
+    SENDLE = "Sendle"
+    """Sendle"""

shipengine_sdk/models/enums/carriers/carriers.py

@@ -0,0 +1,63 @@
+"""Enumeration of valid carrier providers."""
+from enum import Enum
+
+
+class Carriers(Enum):
+    """An enumeration of valid carrier providers."""
+
+    FEDEX = "fedex"
+    """FedEx - Federal Express"""
+
+    UPS = "ups"
+    """UPS - United Parcel Service"""
+
+    USPS = "usps"
+    """USPS - United State Postal Service"""
+
+    STAMPS_COM = "stamps_com"
+    """USPS services via Stamps.com"""
+
+    DHL_EXPRESS = "dhl_express"
+    """DHL Express"""
+
+    DHL_GLOBAL_MAIL = "dhl_global_mail"
+    """DHL ECommerce"""
+
+    CANADA_POST = "canada_post"
+    """Canada Post"""
+
+    AUSTRALIA_POST = "australia_post"
+    """Australia Post"""
+
+    FIRSTMILE = "firstmile"
+    """First Mile"""
+
+    ASENDIA = "asendia"
+    """Asendia"""
+
+    ONTRAC = "ontrac"
+    """OnTrac"""
+
+    APC = "apc"
+    """APC"""
+
+    NEWGISTICS = "newgistics"
+    """Newgistics"""
+
+    GLOBEGISTICS = "globegistics"
+    """Globegistics"""
+
+    RR_DONNELLEY = "rr_donnelley"
+    """RR Donnell"""
+
+    IMEX = "imex"
+    """IMEX"""
+
+    ACCESS_WORLDWIDE = "access_worldwide"
+    """Access Worldwide"""
+
+    PUROLATOR_CA = "purolator_ca"
+    """Purolator Canada"""
+
+    SENDLE = "sendle"
+    """Sendle"""

shipengine_sdk/services/get_carrier_accounts.py

@@ -0,0 +1,56 @@
+"""
+Fetch the carrier account connected to a given ShipEngine Account
+based on the API Key passed into the ShipEngine SDK.
+"""
+from typing import Dict, List, Optional
+
+from ..jsonrpc import rpc_request
+from ..models import CarrierAccount, RPCMethods
+from ..shipengine_config import ShipEngineConfig
+
+
+class GetCarrierAccounts:
+    cached_accounts: List = list()
+
+    def fetch_carrier_accounts(
+        self, config: ShipEngineConfig, carrier_code: Optional[str] = None
+    ) -> List[Dict[str, any]]:
+        if carrier_code is not None:
+            api_response = rpc_request(
+                method=RPCMethods.LIST_CARRIERS.value,
+                config=config,
+                params={"carrierCode": carrier_code},
+            )
+        else:
+            api_response = rpc_request(
+                method=RPCMethods.LIST_CARRIERS.value,
+                config=config,
+            )
+
+        accounts = api_response["result"]["carrierAccounts"]
+        self.cached_accounts = list()
+        for account in accounts:
+            carrier_account = CarrierAccount(account).to_dict()
+            self.cached_accounts.append(carrier_account)
+
+        return self.cached_accounts
+
+    def fetch_cached_carrier_accounts(
+        self, config: ShipEngineConfig, carrier_code: Optional[str]
+    ) -> List[Dict[str, any]]:
+        accounts = self.cached_accounts
+        return (
+            accounts
+            if len(self.cached_accounts) > 0
+            else self.fetch_carrier_accounts(config=config, carrier_code=carrier_code)
+        )
+
+    def get_cached_accounts_by_carrier_code(self, carrier_code: Optional[str]):
+        accounts = list()
+        if carrier_code is not None:
+            return self.cached_accounts
+        else:
+            for account in self.cached_accounts:
+                if account.carrier.code == carrier_code:
+                    accounts.append(account)
+            return accounts

shipengine_sdk/shipengine.py

@@ -1,8 +1,10 @@
 """The entrypoint to the ShipEngine API SDK."""
-from typing import Dict, Union
+from typing import Dict, List, Optional, Union
 
+from .models import CarrierAccount
 from .models.address import Address, AddressValidateResult
 from .services.address_validation import normalize, validate
+from .services.get_carrier_accounts import GetCarrierAccounts
 from .shipengine_config import ShipEngineConfig
 
 
@@ -47,3 +49,11 @@ def normalize_address(
         """Normalize a given address into a standardized format used by carriers."""
         config: ShipEngineConfig = self.config.merge(new_config=config)
         return normalize(address=address, config=config)
+
+    def get_carrier_accounts(
+        self, carrier_code: Optional[str] = None, config: Optional[Dict[str, any]] = None
+    ) -> List[CarrierAccount]:
+        """Fetch a list of the carrier accounts connected to your ShipEngine Account."""
+        config: ShipEngineConfig = self.config.merge(new_config=config)
+        get_accounts = GetCarrierAccounts()
+        return get_accounts.fetch_carrier_accounts(config=config, carrier_code=carrier_code)