Source code for oscarcch.calculator

from datetime import datetime
from decimal import Decimal
from django_statsd.clients import statsd
from . import exceptions, settings
import logging
import requests
import soap

logger = logging.getLogger(__name__)

POSTCODE_LEN = 5
PLUS4_LEN = 4


[docs]class CCHTaxCalculator(object): """ Simple interface between Python and the CCH Sales Tax Office SOAP API. """ precision = settings.CCH_PRECISION wsdl = settings.CCH_WSDL entity_id = settings.CCH_ENTITY divsion_id = settings.CCH_DIVISION max_retries = settings.CCH_MAX_RETRIES
[docs] def estimate_taxes(self, basket, shipping_address): """ DEPRECATED. Use :func:`CCHTaxCalculator.apply_taxes <oscarcch.calculator.CCHTaxCalculator.apply_taxes>` instead. """ statsd.incr('cch.estimate') self.apply_taxes(basket, shipping_address) return basket
[docs] def apply_taxes(self, basket, shipping_address, ignore_cch_fail=False): """ Apply taxes to a Basket instance using the given shipping address. Pass return value of this method to :func:`OrderTaxation.save_details <oscarcch.models.OrderTaxation.save_details>` to persist the taxation details, CCH transaction ID, etc in the database. :param basket: :class:`Basket <oscar.apps.basket.models.Basket>` instance :param shipping_address: :class:`ShippingAddress <oscar.apps.order.models.ShippingAddress>` instance :param ignore_cch_fail: When `True`, allows CCH to fail silently :return: SOAP Response. """ with statsd.timer('cch.apply-time'): response = self._get_response(basket, shipping_address, ignore_cch_fail) if not ignore_cch_fail: self._check_response_messages(response) # Apply taxes to line items for line in basket.all_lines(): line_id = str(line.id) taxes = None if response and response.LineItemTaxes: try: taxes = next(filter(lambda item: item.ID == line_id, response.LineItemTaxes.LineItemTax)) except StopIteration: pass # Taxes come in two forms: quantity and percentage based # We need to handle both of those here. The tricky part is that CCH returns data # for an entire line item (inclusive quantity), but Oscar needs the tax info for # each unit in the line (exclusive quantity). So, we use the details provided to # derive the per-unit taxes before applying them. line.purchase_info.price.clear_taxes() if taxes: for tax in taxes.TaxDetails.TaxDetail: unit_fee = Decimal(str(tax.FeeApplied)) / line.quantity unit_tax = Decimal(str(tax.TaxApplied)) / line.quantity line.purchase_info.price.add_tax( authority_name=tax.AuthorityName, tax_name=tax.TaxName, tax_applied=unit_tax, fee_applied=unit_fee) # Check our work and make sure the total we arrived at matches the total CCH gave us total_line_tax = (line.purchase_info.price.tax * line.quantity).quantize(self.precision) total_applied_tax = Decimal(taxes.TotalTaxApplied).quantize(self.precision) if total_applied_tax != total_line_tax: statsd.incr('cch.miscalculation') raise RuntimeError(( "Taxation miscalculation occurred! " "Details sum to %s, which doesn't match given sum of %s" ) % (total_line_tax, taxes.TotalTaxApplied)) else: line.purchase_info.price.tax = Decimal('0.00') return response
def _get_response(self, basket, shipping_address, ignore_cch_fail=False): """Fetch CCH tax data for the given basket and shipping address""" response = None retry_count = 0 while response is None and retry_count <= self.max_retries: response = self._get_response_inner(basket, shipping_address, ignore_cch_fail, retry_count=retry_count) retry_count += 1 if response: statsd.incr('cch.apply-success') else: statsd.incr('cch.apply-failure') return response def _get_response_inner(self, basket, shipping_address, ignore_cch_fail, retry_count): # Attempt to get a response response = None try: order = self._build_order(basket, shipping_address) response = self.client.service.CalculateRequest(self.entity_id, self.divsion_id, order) # Timeouts (read or connect) get retried except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e: if retry_count >= self.max_retries and not ignore_cch_fail: raise e # Catch any other possible exceptions and, if we're supposed to, ignore them. except Exception as e: if not ignore_cch_fail: raise e # Return our (apparently successful) response return response def _check_response_messages(self, response): """Raise an exception if response messages contains any reported errors.""" if response and response.Messages: for message in response.Messages.Message: if message.Code > 0: raise exceptions.build(message.Severity, message.Code, message.Info) @property def client(self): """Lazy constructor for SOAP client""" return soap.get_client(self.wsdl, 'CCH') def _build_order(self, basket, shipping_address): """Convert an Oscar Basket and ShippingAddresss into a CCH Order object""" order = self.client.factory.create('ns15:Order') order.InvoiceDate = datetime.now(settings.CCH_TIME_ZONE) order.SourceSystem = settings.CCH_SOURCE_SYSTEM order.TestTransaction = settings.CCH_TEST_TRANSACTIONS order.TransactionType = settings.CCH_TRANSACTION_TYPE order.CustomerType = settings.CCH_CUSTOMER_TYPE order.ProviderType = settings.CCH_PROVIDER_TYPE order.TransactionID = 0 order.finalize = settings.CCH_FINALIZE_TRANSACTION for line in basket.all_lines(): qty = getattr(line, 'cch_quantity', line.quantity) if qty <= 0: continue item = self.client.factory.create('ns11:LineItem') item.ID = line.id item.AvgUnitPrice = (line.line_price_excl_tax_incl_discounts / qty).quantize(Decimal('0.00001')) item.Quantity = qty item.ExemptionCode = None item.SKU = self._get_product_data('sku', line) item.ProductInfo = self.client.factory.create('ns21:ProductInfo') item.ProductInfo.ProductGroup = self._get_product_data('group', line) item.ProductInfo.ProductItem = self._get_product_data('item', line) item.NexusInfo = self.client.factory.create('ns14:NexusInfo') item.NexusInfo.ShipFromAddress = self.client.factory.create('ns0:Address') warehouse = line.stockrecord.partner.primary_address if warehouse: item.NexusInfo.ShipFromAddress.Line1 = warehouse.line1 item.NexusInfo.ShipFromAddress.Line2 = warehouse.line2 item.NexusInfo.ShipFromAddress.City = warehouse.city item.NexusInfo.ShipFromAddress.StateOrProvince = warehouse.state postcode, plus4 = self.format_postcode(warehouse.postcode) item.NexusInfo.ShipFromAddress.PostalCode = postcode item.NexusInfo.ShipFromAddress.Plus4 = plus4 item.NexusInfo.ShipFromAddress.CountryCode = warehouse.country.code item.NexusInfo.ShipToAddress = self.client.factory.create('ns0:Address') item.NexusInfo.ShipToAddress.Line1 = shipping_address.line1 item.NexusInfo.ShipToAddress.Line2 = shipping_address.line2 item.NexusInfo.ShipToAddress.City = shipping_address.city item.NexusInfo.ShipToAddress.StateOrProvince = shipping_address.state postcode, plus4 = self.format_postcode(shipping_address.postcode) item.NexusInfo.ShipToAddress.PostalCode = postcode item.NexusInfo.ShipToAddress.Plus4 = plus4 item.NexusInfo.ShipToAddress.CountryCode = shipping_address.country.code order.LineItems.LineItem.append(item) return order def _get_product_data(self, key, line): key = 'cch_product_%s' % key sku = getattr(settings, key.upper()) sku = getattr(line.product.attr, key.lower(), sku) return sku def format_postcode(self, raw_postcode): postcode, plus4 = raw_postcode[:POSTCODE_LEN], None # set Plus4 if PostalCode provided as 9 digits separated by hyphen if len(raw_postcode) == POSTCODE_LEN + PLUS4_LEN + 1: plus4 = raw_postcode[POSTCODE_LEN + 1:] return postcode, plus4