# pylint: disable=unused-argument
from decimal import Decimal as D
from django.conf import settings
from django.contrib.sites.models import Site
from django.db import transaction
from django.utils.translation import gettext_lazy as _
from oscar.apps.order.signals import order_placed
from oscar.core.loading import get_class, get_model
from . import exceptions
Order = get_model("order", "Order")
Line = get_model("order", "Line")
OrderDiscount = get_model("order", "OrderDiscount")
OrderLineDiscount = get_model("order", "OrderLineDiscount")
CommunicationEvent = get_model("order", "CommunicationEvent")
CommunicationEventType = get_model("communication", "CommunicationEventType")
Dispatcher = get_class("communication.utils", "Dispatcher")
Surcharge = get_model("order", "Surcharge")
[docs]class OrderNumberGenerator(object):
"""
Simple object for generating order numbers.
We need this as the order number is often required for payment
which takes place before the order model has been created.
"""
[docs] def order_number(self, basket):
"""
Return an order number for a given basket
"""
return 100000 + basket.id
[docs]class OrderCreator(object):
"""
Places the order by writing out the various models
"""
[docs] def place_order(
self,
basket,
total,
shipping_method,
shipping_charge,
user=None,
shipping_address=None,
billing_address=None,
order_number=None,
status=None,
request=None,
surcharges=None,
**kwargs
):
"""
Placing an order involves creating all the relevant models based on the
basket and session data.
"""
if basket.is_empty:
raise ValueError(_("Empty baskets cannot be submitted"))
if not order_number:
generator = OrderNumberGenerator()
order_number = generator.order_number(basket)
if not status and hasattr(settings, "OSCAR_INITIAL_ORDER_STATUS"):
status = getattr(settings, "OSCAR_INITIAL_ORDER_STATUS")
if Order._default_manager.filter(number=order_number).exists():
raise ValueError(
_("There is already an order with number %s") % order_number
)
with transaction.atomic():
kwargs["surcharges"] = surcharges
# Ok - everything seems to be in order, let's place the order
order = self.create_order_model(
user,
basket,
shipping_address,
shipping_method,
shipping_charge,
billing_address,
total,
order_number,
status,
request,
**kwargs
)
for voucher in basket.vouchers.select_for_update():
if not voucher.is_active(): # basket ignores inactive vouchers
basket.vouchers.remove(voucher)
else:
available_to_user, msg = voucher.is_available_to_user(user=user)
if not available_to_user:
raise ValueError(msg)
# Record any discounts associated with this order
for application in basket.offer_applications:
# Trigger any deferred benefits from offers and capture the
# resulting message
application["message"] = application["offer"].apply_deferred_benefit(
basket, order, application
)
# Record offer application results
if application["result"].affects_shipping:
# Skip zero shipping discounts
shipping_discount = shipping_method.discount(basket)
if shipping_discount <= D("0.00"):
continue
# If a shipping offer, we need to grab the actual discount off
# the shipping method instance, which should be wrapped in an
# OfferDiscount instance.
application["discount"] = shipping_discount
self.create_discount_model(order, application)
self.record_discount(application)
for voucher in basket.vouchers.all():
self.record_voucher_usage(order, voucher, user)
for line in basket.all_lines():
self.create_line_models(order, line)
self.update_stock_records(line)
# Send signal for analytics to pick up
order_placed.send(sender=self, order=order, user=user)
return order
[docs] def create_order_model(
self,
user,
basket,
shipping_address,
shipping_method,
shipping_charge,
billing_address,
total,
order_number,
status,
request=None,
surcharges=None,
**extra_order_fields
):
"""Create an order model."""
order_data = {
"basket": basket,
"number": order_number,
"currency": total.currency,
"total_incl_tax": total.incl_tax,
"total_excl_tax": total.excl_tax,
"shipping_incl_tax": shipping_charge.incl_tax,
"shipping_excl_tax": shipping_charge.excl_tax,
"shipping_tax_code": shipping_charge.tax_code,
"shipping_method": shipping_method.name,
"shipping_code": shipping_method.code,
}
if shipping_address:
order_data["shipping_address"] = shipping_address
if billing_address:
order_data["billing_address"] = billing_address
if user and user.is_authenticated:
order_data["user_id"] = user.id
if status:
order_data["status"] = status
if extra_order_fields:
order_data.update(extra_order_fields)
if "site" not in order_data:
order_data["site"] = Site._default_manager.get_current(request)
order = Order(**order_data)
order.save()
if surcharges is not None:
for charge in surcharges:
Surcharge.objects.create(
order=order,
name=charge.surcharge.name,
code=charge.surcharge.code,
excl_tax=charge.price.excl_tax,
incl_tax=charge.price.incl_tax,
tax_code=charge.price.tax_code,
)
return order
[docs] def create_line_models(self, order, basket_line, extra_line_fields=None):
"""
Create the batch line model.
You can set extra fields by passing a dictionary as the
extra_line_fields value
"""
product = basket_line.product
stockrecord = basket_line.stockrecord
if not stockrecord:
raise exceptions.UnableToPlaceOrder(
"Basket line #%d has no stockrecord" % basket_line.id
)
partner = stockrecord.partner
line_data = {
"order": order,
# Partner details
"partner": partner,
"partner_name": partner.name,
"partner_sku": stockrecord.partner_sku,
"stockrecord": stockrecord,
# Product details
"product": product,
"title": product.get_title(),
"upc": product.upc,
"quantity": basket_line.quantity,
# Price details
"line_price_excl_tax": basket_line.line_price_excl_tax_incl_discounts,
"line_price_incl_tax": basket_line.line_price_incl_tax_incl_discounts,
"line_price_before_discounts_excl_tax": basket_line.line_price_excl_tax,
"line_price_before_discounts_incl_tax": basket_line.line_price_incl_tax,
# Reporting details
"unit_price_incl_tax": basket_line.unit_price_incl_tax,
"unit_price_excl_tax": basket_line.unit_price_excl_tax,
"tax_code": basket_line.tax_code,
}
extra_line_fields = extra_line_fields or {}
if hasattr(settings, "OSCAR_INITIAL_LINE_STATUS"):
if not (extra_line_fields and "status" in extra_line_fields):
extra_line_fields["status"] = getattr(
settings, "OSCAR_INITIAL_LINE_STATUS"
)
if extra_line_fields:
line_data.update(extra_line_fields)
order_line = Line._default_manager.create(**line_data)
self.create_line_price_models(order, order_line, basket_line)
self.create_line_attributes(order, order_line, basket_line)
self.create_line_discount_models(order, order_line, basket_line)
self.create_additional_line_models(order, order_line, basket_line)
return order_line
[docs] def update_stock_records(self, line):
"""
Update any relevant stock records for this order line
"""
if line.product.get_product_class().track_stock:
line.stockrecord.allocate(line.quantity)
def create_line_discount_models(self, order, order_line, basket_line):
for discount in basket_line.discounts:
order_discount = order.discounts.filter(offer_id=discount.offer.id).first()
# If we are unable to find the discount we do not care, the total amount is still saved on the discount model,
# these models are only created so we know how much discount was given per line for each offer
if order_discount:
order_line.discounts.create(
order_discount=order_discount,
is_incl_tax=discount.incl_tax,
amount=discount.amount,
)
[docs] def create_additional_line_models(self, order, order_line, basket_line):
"""
Empty method designed to be overridden.
Some applications require additional information about lines, this
method provides a clean place to create additional models that
relate to a given line.
"""
return
[docs] def create_line_price_models(self, order, order_line, basket_line):
"""
Creates the batch line price models
"""
breakdown = basket_line.get_price_breakdown()
for price_incl_tax, price_excl_tax, quantity in breakdown:
order_line.prices.create(
order=order,
quantity=quantity,
price_incl_tax=price_incl_tax,
price_excl_tax=price_excl_tax,
tax_code=basket_line.tax_code,
)
# pylint: disable=unused-argument
[docs] def create_line_attributes(self, order, order_line, basket_line):
"""
Creates the batch line attributes.
"""
for attr in basket_line.attributes.all():
order_line.attributes.create(
option=attr.option, type=attr.option.code, value=attr.value
)
[docs] def create_discount_model(self, order, discount):
"""
Create an order discount model for each offer application attached to
the basket.
"""
order_discount = OrderDiscount(
order=order,
message=discount["message"] or "",
offer_id=discount["offer"].id,
frequency=discount["freq"],
amount=discount["discount"],
)
result = discount["result"]
if result.affects_shipping:
order_discount.category = OrderDiscount.SHIPPING
elif result.affects_post_order:
order_discount.category = OrderDiscount.DEFERRED
voucher = discount.get("voucher", None)
if voucher:
order_discount.voucher_id = voucher.id
order_discount.voucher_code = voucher.code
order_discount.save()
def record_discount(self, discount):
discount["offer"].record_usage(discount)
if "voucher" in discount and discount["voucher"]:
discount["voucher"].record_discount(discount)
[docs] def record_voucher_usage(self, order, voucher, user):
"""
Updates the models that care about this voucher.
"""
voucher.record_usage(order, user)
[docs]class OrderDispatcher:
"""
Dispatcher to send concrete order related emails.
"""
# Event codes
ORDER_PLACED_EVENT_CODE = "ORDER_PLACED"
def __init__(self, logger=None, mail_connection=None):
self.dispatcher = Dispatcher(logger=logger, mail_connection=mail_connection)
[docs] def dispatch_order_messages(
self, order, messages, event_code, attachments=None, **kwargs
):
"""
Dispatch order-related messages to the customer.
"""
self.dispatcher.logger.info(
"Order #%s - sending %s messages", order.number, event_code
)
if order.is_anonymous:
email = kwargs.get("email_address", order.guest_email)
dispatched_messages = self.dispatcher.dispatch_anonymous_messages(
email, messages, attachments
)
else:
dispatched_messages = self.dispatcher.dispatch_user_messages(
order.user, messages, attachments
)
try:
event_type = CommunicationEventType.objects.get(code=event_code)
except CommunicationEventType.DoesNotExist:
event_type = None
self.create_communication_event(order, event_type, dispatched_messages)
[docs] def create_communication_event(self, order, event_type, dispatched_messages):
"""
Create order communications event for audit.
"""
if dispatched_messages and event_type is not None:
CommunicationEvent._default_manager.create(
order=order, event_type=event_type
)
def send_order_placed_email_for_user(self, order, extra_context, attachments=None):
event_code = self.ORDER_PLACED_EVENT_CODE
messages = self.dispatcher.get_messages(event_code, extra_context)
self.dispatch_order_messages(
order, messages, event_code, attachments=attachments
)