Repository URL to install this package:
|
Version:
0.4.139 ▾
|
lib-py-b2b
/
order_builder.py
|
|---|
import abc
from lib_b2b.additional_charge import AdditionalCharge
from lib_b2b.address import Address
from lib_b2b.discount import Discount
from lib_b2b.order import OrderLine
from lib_b2b.order_request import OrderRequest, ShopifyOrderRequest
from lib_b2b.order_status import OrderStatus
import logging
from os import environ
from lib_b2b.order_totals import OrderTotals
from lib_b2b.refund import Refund
log_level = logging.getLevelName(environ['LOG_LEVEL']) if 'LOG_LEVEL' in environ else logging.INFO
logging.basicConfig(format='%(asctime)s,%(msecs)d %(levelname)-8s [%(filename)s:%(lineno)d] %(message)s',
datefmt='%d-%m-%Y:%H:%M:%S',
level=log_level)
logger = logging.getLogger('lib-b2b-order-builder')
class OrderBuilder:
@staticmethod
def for_(request: (OrderRequest, ShopifyOrderRequest)) -> 'OrderBuilder':
if isinstance(request, ShopifyOrderRequest):
from lib_b2b.shopify_order_builder import ShopifyOrderBuilder
return ShopifyOrderBuilder(request)
else:
from lib_b2b.standard_order_builder import StandardOrderBuilder
return StandardOrderBuilder(request)
def __init__(self, request: OrderRequest):
self.request = request
self._customer = None
def build(self):
_order = {}
_order.update(self._build_standard_attributes())
_order.update(self._build_identifiers())
_order.update(self._build_dates())
_order.update(self._build_notification_urls())
_ship_to = self._build_shipping_address()
_discounts = self._build_discounts()
_order_lines, _order_line_additional_charges = self._build_lines(ship_to=_ship_to)
_additional_charges = _order_line_additional_charges
_additional_charges.extend(self._build_additional_charges())
_refunds = self._build_refunds()
_totals = self._build_totals(lines=_order_lines, discounts=_discounts,
additional_charges=_additional_charges, refunds=_refunds)
_order.update({'shipTo': _ship_to.as_dict()})
_order.update({'order_lines': list(map(lambda x: x.as_dict(), _order_lines))})
_order.update({'additional_charges': list(map(lambda x: x.as_dict(), _additional_charges))})
_order.update({'discounts': list(map(lambda x: x.as_dict(), _discounts))})
_order.update(_totals.as_dict())
return _order
@property
@abc.abstractmethod
def customer(self):
raise NotImplementedError
@property
@abc.abstractmethod
def order_id(self):
raise NotImplementedError
def _build_standard_attributes(self) -> dict:
_order = {}
_order['order_status'] = OrderStatus.RECEIVED.value
_order['id'] = self.order_id
_order['customer_mode'] = self.customer['mode']
_order['integration_type'] = self.customer.get('integration_type', 'standard')
return _order
def _build_shipping_address(self) -> Address:
return self.request.dialect.shipTo
@abc.abstractmethod
def _build_identifiers(self) -> dict:
raise NotImplementedError
@abc.abstractmethod
def _build_dates(self) -> dict:
raise NotImplementedError
@abc.abstractmethod
def _build_notification_urls(self) -> dict:
raise NotImplementedError
@abc.abstractmethod
def _build_lines(self, ship_to: Address) -> ([OrderLine], [AdditionalCharge]):
raise NotImplementedError
@abc.abstractmethod
def _build_additional_charges(self) -> [AdditionalCharge]:
raise NotImplementedError
@abc.abstractmethod
def _build_discounts(self) -> [Discount]:
raise NotImplementedError
@abc.abstractmethod
def _build_refunds(self) -> [Refund]:
raise NotImplementedError
@abc.abstractmethod
def _build_totals(self, lines: [OrderLine], discounts: [Discount],
additional_charges: [AdditionalCharge], refunds: [Refund]) -> OrderTotals:
raise NotImplementedError