Repository URL to install this package:
|
Version:
0.4.139 ▾
|
lib-py-b2b
/
erp.py
|
|---|
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