Skip to content

Instantly share code, notes, and snippets.

@portedison

portedison/dynamics~abstractions.py

Last active August 23, 2024 04:33
Show Gist options
  • Save portedison/a5f426979b4888d89a38c292ad063ed0 to your computer and use it in GitHub Desktop.
Save portedison/a5f426979b4888d89a38c292ad063ed0 to your computer and use it in GitHub Desktop.
Download ZIP
dynamics / abstractions
Raw
dynamics~abstractions.py
# This is an example of some mixins I created for a Python / Django cart,
# these mixins ca be applied like so -
# - Cart(AbstractItemCollection, DynamicsPriceValidatedMixin):
# - CartLineItem(AbstractLineItem, DynamicsPriceMixin):
#
# The cart required unique prices per user on selected items, these prices
# can update at any time, so I revalidate the items -
# - when hitting the payment page
# - when we call get_cart / get_checkout (not is_order)
# - a. last_price_revalidation is greater than
# - b. when one of the items in the cart has a revalidate state
#
# I use concurrent requests to process multiple requests for efficiency
import concurrent.futures
import logging
from datetime import timedelta
from decimal import ROUND_HALF_UP, Decimal
from functools import partial
from django.db import models
from django.utils import timezone
from customers.utils import get_current_company
from dynamics365.settings import DEFAULT_SAMPLE_PRICE
from ecommerce.settings import DECIMAL_MAX_DIGITS, DECIMAL_PLACES
logger = logging.getLogger(__name__)
def log_error(message):
message = '{}: {}'.format(
timezone.now().strftime("%Y-%m-%d %H:%M:%S"), message)
print(message)
logger.error(message)
"""
Example, price response
[{
'$id': '1',
'NetAmount': 0.0,
'Currency': '',
'Quantity': 1.0,
'Price': 0.0,
'PriceUnit': 1.0,
'PriceUOM': 'EA',
'DiscountAmount': 0.0,
'DiscountPercentage1': 0.0,
'DiscountPercentage2': 0.0,
'Markup': 0.0,
'SalesPrice': 0.0,
'SalesUOM': 'EA',
'SiteId': '',
'WarehouseId': '',
'PriceFromDate': '1900-01-01T12:00:00',
'PriceToDate': '1900-01-01T12:00:00',
'DiscountFromDate': '1900-01-01T12:00:00',
'DiscountToDate': '1900-01-01T12:00:00'
}]
"""
class DynamicsPriceValidatedMixin(models.Model):
"""
bang this on the cart / checkout, this will allow you to revalidate,
any of the line items that depend on a unique price per user, you can call
revalidate_prices() whereever / whenever you see fit, at present this is
- when hitting the payment page
- when we call get_cart / get_checkout (not is_order)
- a. last_price_revalidation is greater than
- b. when one of the items in the cart has a revalidate state
it's important to note that we need the company to be evaluated
while we still have context, otherwise we get None, eg. in a parallel job
"""
MAXIMUM_ELAPSED_TIME = 60
last_price_revalidation = models.DateTimeField(blank=True, null=True)
class Meta:
abstract = True
def get_company_from_parent_user(self):
company = get_current_company()
if not company and self.user:
current_company_for_user = \
self.user.user_companies.filter(company__is_live=True).first()
if current_company_for_user:
company = current_company_for_user.company
return company
@property
def price_is_locked(self):
return bool(hasattr(self, 'order') and self.order.is_order)
@property
def requires_price_revalidation(self):
if self.price_is_locked:
return False
return any((
# 1. not validated or time elapsed
not self.last_price_revalidation or self.last_price_revalidation <
timezone.now() - timedelta(minutes=self.MAXIMUM_ELAPSED_TIME),
# 2. contains invalid items
any((li.requires_price_revalidation for li in self.line_items.all()))))
def revalidate_prices(self):
company = self.get_company_from_parent_user()
jobs = {}
revalidated_lines = {}
for line_item in self.line_items.all():
# need a check here for old orders (items no longer exist)
if line_item.item and line_item.item.sku and \
isinstance(line_item, DynamicsPriceMixin):
jobs.update({
line_item.item.sku:
partial(line_item.revalidate_price, company)
})
with concurrent.futures.ThreadPoolExecutor() as executor:
running = dict((executor.submit(func), label)
for label, func in jobs.items())
for future in concurrent.futures.as_completed(running):
if future.exception() is not None:
raise future.exception()
result = future.result()
if result:
revalidated_lines[running[future]] = result
self.last_price_revalidation = timezone.now()
self.save()
return self
class DynamicsPriceMixin(models.Model):
"""
bang this onto your CartLineItem, OrderLineItem, then when required
we can get / set the price for user
if it's a checkout we need to also reset the price_each / total
ensure that all AbstractCartItem point back to this e.g.
def cart_price(self, line_item):
return line_item.price_for_user
def valid_for_line_item(self, line_item):
...
if line_item.price_for_user_status == line_item.PRICE_STATUS_PENDING:
return (False, [PRICE_PENDING_MESSAGE])
...
def use_price_for_user(self, user, company):
...
def user_has_purchase_permission(self, user, company):
...
TODOPIET - when variant is added
- use get_sample_pricing, get_cut_length_pricing, get_roll_pricing
just using get_cut_length_pricing atm
- handle invalid status = no price - dont try to revalidate
"""
PRICE_STATUS_PENDING = 'pending'
PRICE_STATUS_INVALID = 'invalid'
PRICE_STATUS_FAILED = 'failed'
PRICE_STATUS_UNAUTHORISED = 'unauthorised'
PRICE_STATUS_VALID = 'valid'
PRICE_STATUS_NA = 'na'
PRICE_STATUS_CHOICES = (
(PRICE_STATUS_PENDING, 'Pending'),
(PRICE_STATUS_FAILED, 'Failed'),
(PRICE_STATUS_INVALID, 'Invalid'),
(PRICE_STATUS_UNAUTHORISED, 'Unauthorised'),
(PRICE_STATUS_VALID, 'Valid'),
(PRICE_STATUS_NA, 'N/A'),
)
PRICE_STATUS_FORCE_REVALIDATE = [
PRICE_STATUS_PENDING,
PRICE_STATUS_FAILED,
]
price_for_user = models.DecimalField(
max_digits=DECIMAL_MAX_DIGITS, decimal_places=DECIMAL_PLACES,
default=Decimal(0))
price_for_user_status = models.CharField(max_length=30,
choices=PRICE_STATUS_CHOICES, default=PRICE_STATUS_PENDING)
class Meta:
abstract = True
@property
def price_is_locked(self):
return bool(hasattr(self, 'order') and self.order.is_order)
@property
def requires_price_revalidation(self):
if self.price_is_locked:
return False
return any((
# 1. pending / failed
self.price_for_user_status in self.PRICE_STATUS_FORCE_REVALIDATE,
# 2. unauthorised - but permission has changed
(self.price_for_user_status == self.PRICE_STATUS_UNAUTHORISED and
self.item.user_has_purchase_permission(
self.parent.user,
self.parent.get_company_from_parent_user())),
# 2. valid - but permission has changed
(self.price_for_user_status == self.PRICE_STATUS_VALID and
not self.item.user_has_purchase_permission(
self.parent.user,
self.parent.get_company_from_parent_user()))))
def revalidate_price(self, company=None):
if not self.item:
return self
company = company or self.parent.get_company_from_parent_user()
# some items don't require us the validate
# in that case just skip the process
if self.price_for_user_status in [
self.PRICE_STATUS_NA, self.PRICE_STATUS_INVALID] or \
self.price_is_locked:
return self
# user does not have permission to purchase this product
if not self.item.user_has_purchase_permission(
self.parent.user,
self.parent.get_company_from_parent_user()):
self.price_for_user_status = self.PRICE_STATUS_UNAUTHORISED
self.save()
return self
# some products don't require us to validate
if not self.item.use_price_for_user(
self.parent.user,
self.parent.get_company_from_parent_user()):
self.price_for_user_status = self.PRICE_STATUS_NA
self.save()
return self
try:
price_data = self.item.api_client.get_cut_length_pricing(
self.item.sku, company.dynamics_id)
price_data = price_data[0]
self.price_for_user = price_data.get('Price') or DEFAULT_SAMPLE_PRICE
self.price_for_user_status = self.PRICE_STATUS_VALID
if isinstance(self.price_each, models.DecimalField) and \
isinstance(self.total, models.DecimalField):
self.price_each = self.recalculate_price_each()
self.total = self.recalculate_total()
except IndexError as e:
self.price_for_user_status = self.PRICE_STATUS_FAILED
message = 'revalidate_price ({}): {}'.format(
self.item.id, {'error': str(e)})
log_error(message)
except Exception as e:
self.price_for_user_status = self.PRICE_STATUS_FAILED
message = 'revalidate_price ({}): {}'.format(
self.item.id, {'error': str(e)})
log_error(message)
self.save()
return self
def recalculate_price_each(self):
if not self.item:
return None
item_total = self.item.price_for_user
property_total = sum(
lp.price_each for lp in self.valid_line_properties)
total = Decimal(item_total) + Decimal(property_total)
return Decimal(total.quantize(Decimal('.01'), rounding=ROUND_HALF_UP))
def recalculate_total(self):
if not self.item:
return None
total = self.price_each * self.quantity
return Decimal(total.quantize(Decimal('.01'), rounding=ROUND_HALF_UP))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment