Why Gemfury? Push, build, and install  RubyGems npm packages Python packages Maven artifacts PHP packages Go Modules Debian packages RPM packages NuGet packages

Repository URL to install this package:

Details    
lib-py-b2b / erp.py
Size: Mime:
import abc
from abc import ABCMeta
from decimal import Decimal

from lib_b2b.base import BaseClass
from .address import Address
from datetime import datetime


class ERP:
    @staticmethod
    def for_(customer_edi_id: str):
        from lib_b2b.profile import Profile
        from .glovia import Glovia
        profile = Profile.profile_for(customer=customer_edi_id)
        system = profile.integration_config.system
        if system == 'glovia':
            return Glovia()
        else:
            return ERP.default()

    @staticmethod
    def default():
        from .glovia import Glovia
        return Glovia()

    @abc.abstractmethod
    def add(self, order_id: str):
        """
        Add an order to the erp system
        :param order_id: the order identifier
        :type order_id: str
        :raises IntegrationError if unable to add the order into the erp
        :return: None
        :rtype: None
        """
        raise NotImplementedError

    @abc.abstractmethod
    def accept(self, order_id: str):
        """
        Makes any changes to the order that are required after the order has been added and accepted
        :param order_id: the order identifier
        :type order_id: str
        :return: None
        :rtype: None
        """

    @abc.abstractmethod
    def reject(self, order_id: str):
        """
        Handles changes to the erp system that occur as a result of order rejection; potentially removing it.
        :param order_id: the order identifier
        :type order_id: str
        :return: None
        :rtype: None
        """

    @abc.abstractmethod
    def cancel(self, order_id: str):
        """
        Handle modifications to the erp order when an order is cancelled upstream
        :param order_id: the order identifier
        :type order_id: str
        :return: True if the order was cancelled
        :rtype: bool
        :raises OrderCancelError
        """

    @abc.abstractmethod
    def cancel_line(self, order_id: str, order_line_id: str, quantity: int = -1):
        """
        Handle modifications to the erp order when an order line is cancelled upstream
        :param order_id: the order identifier
        :type order_id: str
        :param quantity: the amount to cancel. default it -1 which signifies the entire order line quantity
        :type quantity: int
        :param order_line_id: the order line identifier
        :type order_line_id: str
        :return: True if the line was cancelled
        :rtype: bool
        :raises LineCancelError
        """

    @abc.abstractmethod
    def update_address(self, ccn: str, sales_order: str, new_ship_address: Address, tax_location_code: str):
        """
        Update the address on an existing sales order
        :param ccn: organization
        :type ccn: str
        :param sales_order: order number
        :type sales_order: str
        :param new_ship_address: new address
        :type new_ship_address: Address
        :param tax_location_code: tax location for the new address
        :type tax_location_code: str
        :raises IntegrationError if unable to add the order into the erp
        :return: None
        :rtype: None
        """
        raise NotImplementedError

    @abc.abstractmethod
    def update_ship_date(self, ccn: str, sales_order: str, new_ship_date: datetime):
        """
        Update the ship date on an existing sales order
        :param ccn: organization
        :type ccn: str
        :param sales_order: order number
        :type sales_order: str
        :param new_ship_date: the new date that the order needs to be shipped
        :type new_ship_date: a datetime object
        :raises IntegrationError if unable to add the order into the erp
        :return: None
        :rtype: None
        """
        raise NotImplementedError

    @abc.abstractmethod
    def fetch_order_by_id(self, order_id: str) -> 'ERPOrder':
        """
        Retrieve an ERPOrderLine object from the ERP system
        :param order_id: the order identifier
        :type order_id: str
        :return: an order header object
        :rtype: ERPOrder
        """
        raise NotImplementedError

    @abc.abstractmethod
    def has_drop_shipments(self, order_id: str) -> bool:
        """
        determines whether any of the lines on the order are drop shipments
        :param order_id: the order identifier
        :type order_id: str
        :return: whether the order contains any drop shipments
        :rtype: bool
        """
        raise NotImplementedError

    @abc.abstractmethod
    def fetch_order_lines_by_id(self, order_id: str) -> ['ERPOrderLine']:
        """
        Retrieve ERPOrderLine objects from the ERP system
        :param order_id: the order identifier
        :type order_id: str
        :return: a list of ERPOrderLine objects
        :rtype: [ERPOrderLine]
        """
        raise NotImplementedError

    @abc.abstractmethod
    def fetch_order_line(self, ccn: str, sales_order: str, sales_order_line: str) -> 'ERPOrderLine':
        """
        Retrieve an ERPOrderLine object from the ERP system
        :param ccn: the organization identifier
        :type ccn: str
        :param sales_order: the sales order identifier
        :type sales_order: str
        :param sales_order_line: the sales order line identifier
        :type sales_order_line: str
        :return: an order line object
        :rtype: ERPOrderLine
        """
        raise NotImplementedError

    @abc.abstractmethod
    def fetch_order_line_by_id(self, order_line_id: str) -> 'ERPOrderLine':
        """
        Retrieve an ERPOrderLine object based on the line id
        :param order_line_id: order line identifier
        :type order_line_id: str
        :return: the order line object
        :rtype: ERPOrderLine
        """
        raise NotImplementedError

    @abc.abstractmethod
    def fetch_customer(self, customer_edi_id: str) -> 'ERPCustomer':
        """
        Retrieve the erp system customer record
        :param customer_edi_id: the unique edi id for the customer (buy location)
        :type customer_edi_id: str
        :return: the customer record
        :rtype: ERPCustomer
        """
        raise NotImplementedError

    @abc.abstractmethod
    def find_tax_location_code(self, customer_edi_id: str, ship_to: Address) -> str:
        """
        Find the correct tax location code to use for updating the erp
        :param customer_edi_id:
        :type customer_edi_id:
        :param ship_to:
        :type ship_to:
        :return:
        :rtype:
        """
        raise NotImplementedError

    @abc.abstractmethod
    def find_tax_rates(self, customer_edi_id: str, ship_to: Address) -> ['ERPTaxRate']:
        """
        Find the correct tax location code to use for updating the erp
        :param customer_edi_id:
        :type customer_edi_id:
        :param ship_to:
        :type ship_to:
        :return: list of tax rates that apply of the address, None is the customer is not taxable
        :rtype: ['ERPTaxRate']
        """
        raise NotImplementedError

    @abc.abstractmethod
    def is_valid_item_for_customer(self, customer_edi_id: str, customer_item_name: str) -> bool:
        """
        Determine if the item is a valid item for the customer
        :param customer_edi_id: the customer number
        :type customer_edi_id: str
        :param customer_item_name: the customer defined item name
        :type customer_item_name: str
        :return: boolean represented if the item is valid
        :rtype: bool
        """
        raise NotImplementedError

    @abc.abstractmethod
    def get_customer_price(self, customer_edi_id: str, customer_item_name) -> Decimal:
        """
        Retrieve the customer specific item pricing
        :param customer_edi_id: the customer number
        :type customer_edi_id: str
        :param customer_item_name: the customer defined item name
        :type customer_item_name: str
        :return: the unit price in dollars
        :rtype: decimal.Decimal
        """
        raise NotImplementedError

    @abc.abstractmethod
    def get_customer_item(self, customer_edi_id: str,
                          customer_item_name: str = None, customer_item_revision: str = ' ',
                          item_name: str = None, revision: str = ' ') -> 'ERPItem':
        """
        Retrieve customer specific item information
        :param revision: the revision of the item
        :type revision: str
        :param item_name: the ERP item name
        :type item_name: str
        :param customer_edi_id: the customer number
        :type customer_edi_id: str
        :param customer_item_name: the customer defined item name
        :type customer_item_name: str
        :param customer_item_revision: the revision of the item
        :type customer_item_revision: str
        :return: the item information
        :rtype: ERPItem
        :rtype: ERPItem
        """
        raise NotImplementedError

    @abc.abstractmethod
    def get_customer_items(self, customer_edi_id: str) -> ['ERPItem']:
        """
        Retrieve list of all defined items for the customer
        :param customer_edi_id: the customer number
        :type customer_edi_id: str
        :return: list of the item informationm
        :rtype: [ERPItem]
        """
        raise NotImplementedError

    @abc.abstractmethod
    def get_customer_inventory(self, customer_edi_id: str) -> [('ERPItem', int)]:
        """
        Retrieve list of all defined items for the customer along with their on hand qty
        :param customer_edi_id: the customer number
        :type customer_edi_id: str
        :return: list of the item informationm
        :rtype: [ERPItem]
        """
        raise NotImplementedError

    @abc.abstractmethod
    def get_inventory(self, ccn: str, location: str = None) -> [('ERPItem', int)]:
        """
        Retrieve list of all defined items edi customers along with their on hand qty
        :param location: optional location parameter
        :type location: str
        :param ccn: ccn limitation
        :type ccn: str
        :return: list of the item informationm
        :rtype: [('ERPItem', int)]:
        """
        raise NotImplementedError


class ERPCustomer(BaseClass, metaclass=ABCMeta):
    def __init__(self, ccn: str, customer: str, buy_location: str):
        self.buy_location = buy_location
        self.customer = customer
        self.ccn = ccn


class ERPOrder(BaseClass):
    def __init__(self, sales_ccn: str, sales_order: str, entry_date: datetime,
                 order_date: datetime, customer: str, customer_location: str,
                 edi_code: str, purchase_order: str, required_date: datetime,
                 promise_date: datetime):
        self.promise_date = promise_date
        self.required_date = required_date
        self.purchase_order = purchase_order
        self.edi_code = edi_code
        self.customer_location = customer_location
        self.customer = customer
        self.order_date = order_date
        self.entry_date = entry_date
        self.sales_order = sales_order
        self.sales_ccn = sales_ccn

    @property
    def order_id(self):
        order_id_seg1 = self.edi_code if self.edi_code else self.customer_location
        order_id = f"{order_id_seg1}-{self.purchase_order}".strip()
        return order_id


class ERPOrderLine(BaseClass):

    def __init__(self, customer: str, customer_location: str, edi_code: str,
                 sales_ccn: str, sales_order: str, sales_order_line: str,
                 purchase_order: str, purchase_order_line: str, item: str,
                 item_revision: str, backorder_qty: int, committed_qty: int,
                 cancelled_qty: int, shipped_qty: int, order_qty: int,
                 one_time_ship_to: Address, drop_ship: bool, should_drop_ship: bool,
                 drop_ship_instructions_id: str = None):
        self.should_drop_ship = should_drop_ship
        self.drop_ship_instructions_id = drop_ship_instructions_id
        self.drop_ship = drop_ship
        self.one_time_ship_to = one_time_ship_to
        self.shipped_qty = shipped_qty
        self.order_qty = order_qty
        self.cancelled_qty = cancelled_qty
        self.committed_qty = committed_qty
        self.backorder_qty = backorder_qty
        self.item_revision = item_revision
        self.item = item
        self.purchase_order_line = purchase_order_line
        self.purchase_order = purchase_order
        self.sales_order_line = sales_order_line
        self.sales_order = sales_order
        self.sales_ccn = sales_ccn
        self.edi_code = edi_code
        self.customer_location = customer_location
        self.customer = customer

    @property
    def shipment_id(self):
        shipment_id_seg1 = self.edi_code if self.edi_code else self.customer_location
        shipment_id = f"{shipment_id_seg1}-{self.purchase_order}-{self.purchase_order_line}".strip()
        return shipment_id

    @property
    def order_id(self):
        order_id_seg1 = self.edi_code if self.edi_code else self.customer_location
        order_id = f"{order_id_seg1}-{self.purchase_order}".strip()
        return order_id

    @property
    def order_line_id(self):
        order_line_id_seg1 = self.edi_code if self.edi_code else self.customer_location
        order_line_id = f"{order_line_id_seg1}-{self.purchase_order}-{self.purchase_order_line}".strip()
        return order_line_id

    @property
    def fulfilled(self) -> bool:
        return self.order_qty <= (self.cancelled_qty + self.shipped_qty)

    @property
    def required_qty(self):
        return self.order_qty - self.cancelled_qty - self.shipped_qty


class ERPItem(BaseClass):
    def __init__(self, customer: ERPCustomer, customer_item_name: str, customer_item_revision: str,
                 item: str, item_revision: str, upc: str, accounting_code: str, selling_unit_of_measure: str,
                 stocking_unit_of_measure: str, unit_price: Decimal, sales_tax: bool, excise_tax: bool, use_tax: bool,
                 standard_cost: Decimal, stock_item: bool, primary_vender: str, purchase_location: str,
                 drop_ship_item: bool, height: Decimal, length: Decimal, width: Decimal, weight: Decimal,
                 weight_unit_of_measure: str, product_line: str, product_code: str):
        self.product_code = product_code
        self.use_tax = use_tax
        self.product_line = product_line
        self.weight_unit_of_measure = weight_unit_of_measure
        self.weight = weight
        self.width = width
        self.length = length
        self.height = height
        self.drop_ship_item = drop_ship_item
        self.purchase_location = purchase_location
        self.primary_vender = primary_vender
        self.stock_item = stock_item
        self.standard_cost = standard_cost
        self.excise_tax = excise_tax
        self.sales_tax = sales_tax
        self.unit_price = unit_price
        self.stocking_unit_of_measure = stocking_unit_of_measure
        self.selling_unit_of_measure = selling_unit_of_measure
        self.accounting_code = accounting_code
        self.upc = upc
        self.item_revision = item_revision
        self.item = item
        self.customer_item_revision = customer_item_revision
        self.customer_item_name = customer_item_name
        self.customer = customer


class ERPTaxRate(BaseClass):
    def __init__(self, rate: Decimal, tax_location_code: str, tax_rate_code: str, tax_rate_description: str):
        self.tax_rate_description = tax_rate_description
        self.tax_rate_code = tax_rate_code
        self.tax_location_code = tax_location_code
        self.rate = rate