Repository URL to install this package:
|
Version:
0.4.202 ▾
|
lib-py-b2b
/
standard_order_builder.py
|
|---|
from dateutil.parser import parse
from dateutil.tz import UTC
from lib_b2b.additional_charge import AdditionalCharge
from lib_b2b.address import Address
from lib_b2b.customer import Customer
from lib_b2b.discount import Discount
from lib_b2b.financial_status import FinancialStatus
from lib_b2b.order import Order
from lib_b2b.order_dates import OrderDatesDelegate
from lib_b2b.order_line import OrderLine
from lib_b2b.order_builder import OrderBuilder
from lib_b2b.order_request import OrderRequest
from datetime import datetime, timedelta, timezone
import logging
from os import environ
from lib_b2b.order_totals import OrderTotals
from lib_b2b.refund import Refund
logger = logging.getLogger(__name__)
class StandardOrderBuilder(OrderBuilder):
def __init__(self, request: OrderRequest):
super().__init__(request)
@property
def customer(self):
if not self._customer:
self._customer = Customer.fetch_by_edi_id(self.request.customer_edi_id)
return self._customer
@property
def order_id(self):
return Order.generate_id(self.customer['customer_edi_id'], str(self.request.purchase_order))
def _build_identifiers(self):
_order = {}
_order['channel_order_id'] = self.request.channel_order_id or self.request.purchase_order
_order['channel_name'] = self.request.channel_name or "standard"
_order['customer_edi_id'] = self.request.customer_edi_id
_order['purchase_order'] = self.request.purchase_order
_order['purchase_order_revision'] = self.request.purchase_order_revision or 1
_order['version'] = self.request.purchase_order_revision or 1
_order['carrier'] = self.request.purchase_order_revision or "FDXG"
if self.request.note:
_order['note'] = self.request.note
_order['financial_status'] = FinancialStatus.PAID.value
return _order
def _build_dates(self):
_order = {}
today = datetime.now(tz=timezone.utc)
if self.request.manually_entered:
_order['order_date'] = (today - timedelta(days=1)).isoformat()
else:
_order['order_date'] = today.isoformat()
date_delegate = OrderDatesDelegate(self.customer['customer_edi_id'], parse(_order['order_date']))
release_date = date_delegate.release_date.date
_order['release_on'] = f"{release_date.isoformat()}"
required_date = self.request.not_before_date or (datetime.now().astimezone(UTC) + timedelta(days=2)).isoformat()
_order['required_date'] = required_date
_order['not_before_date'] = self.request.not_before_date
return _order
def _build_notification_urls(self):
return {'notification_urls': self.request.notification_urls}
def _build_lines(self, ship_to: Address) -> ([OrderLine], [AdditionalCharge]):
_lines = []
_addl_chrgs = []
# Orders Lines
line_count = 1
for i, order_line in enumerate(self.request.order_lines, 1):
_line = OrderLine(
purchase_order_line=order_line.get('purchase_order_line', str(line_count).zfill(4)),
product_external_id=order_line.get('product_external_id'),
product_id=order_line.get('product_id'),
quantity=order_line.get('quantity', 1),
note=order_line.get('note', None),
order_id=self.order_id,
customer_edi_id=self.request.customer_edi_id,
ship_to=ship_to,
cancelled=False
)
_lines.append(_line)
if _line.requires_freight(self.request.customer_edi_id):
_acs = AdditionalCharge.create_freight_charges(customer=self.request.customer_edi_id,
order_line=_line,
ship_to=ship_to)
_addl_chrgs.extend(_acs)
line_count += 1
return _lines, _addl_chrgs
def _build_additional_charges(self) -> [AdditionalCharge]:
return []
def _build_discounts(self) -> [Discount]:
return []
def _build_refunds(self) -> [Refund]:
return []
def _build_totals(self, lines: [OrderLine], discounts: [Discount],
additional_charges: [AdditionalCharge], refunds: [Refund]) -> OrderTotals:
return OrderTotals.create(order_lines=lines,
discounts=discounts,
additional_charges=additional_charges,
refunds=refunds)