byceps.services.shop.order.order_domain_service._build_incoming_line_items() - Code Metrics - Inspection of "Persist log entry for placed order together with o..." - byceps/byceps - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( d0a955...ec822a )

by Jochen

created 2024-03-13 00:27 UTC

_build_incoming_line_items() A

↳ Parent: byceps.services.shop.order.order_domain_service

Complexity

Conditions

Size

Total Lines	20
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	3.1852

Importance

Changes

Metric	Value
cc	2
eloc	17
nop	1
dl	0
loc	20
ccs	2
cts	6
cp	0.3333
crap	3.1852
rs	9.55
c	0
b	0
f	0

"""
byceps.services.shop.order.order_domain_service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

:Copyright: 2014-2024 Jochen Kupperschmidt
:License: Revised BSD (see `LICENSE` file for details)
"""

from collections.abc import Iterator
from datetime import datetime, timedelta

from moneyed import Currency, Money

from byceps.events.base import EventUser
from byceps.events.shop import ShopOrderCanceledEvent, ShopOrderPaidEvent
from byceps.services.shop.cart.models import Cart, CartItem
from byceps.services.shop.shop.models import ShopID
from byceps.services.shop.storefront.models import StorefrontID
from byceps.services.user.models.user import User
from byceps.util.result import Err, Ok, Result
from byceps.util.uuid import generate_uuid7

from .errors import (
    CartEmpty,
    OrderAlreadyCanceledError,
    OrderAlreadyMarkedAsPaidError,
)
from .models.checkout import IncomingLineItem, IncomingOrder
from .models.log import OrderLogEntry, OrderLogEntryData
from .models.order import LineItemID, Order, Orderer, OrderID, PaymentState
from .models.payment import AdditionalPaymentData, Payment


OVERDUE_THRESHOLD = timedelta(days=14)


def place_order(
    created_at: datetime,
    shop_id: ShopID,
    storefront_id: StorefrontID,
    orderer: Orderer,
    currency: Currency,
    cart: Cart,
) -> Result[tuple[IncomingOrder, OrderLogEntry], CartEmpty]:
    """Place an order."""
    cart_items = cart.get_items()
    if not cart_items:
        return Err(CartEmpty())

    line_items = list(_build_incoming_line_items(cart_items))

    total_amount = cart.calculate_total_amount()

    processing_required = any(
        line_item.processing_required for line_item in line_items
    )

    incoming_order = IncomingOrder(
        id=OrderID(generate_uuid7()),
        created_at=created_at,
        shop_id=shop_id,
        storefront_id=storefront_id,
        orderer=orderer,
        line_items=line_items,
        total_amount=total_amount,
        processing_required=processing_required,
    )

    log_entry = _build_order_placed_log_entry(incoming_order)

    return Ok((incoming_order, log_entry))


def _build_incoming_line_items(
    cart_items: list[CartItem],
) -> Iterator[IncomingLineItem]:
    """Build incoming line item objects from the cart's content."""
    for cart_item in cart_items:
        article = cart_item.article
        quantity = cart_item.quantity
        line_amount = cart_item.line_amount

        yield IncomingLineItem(
            id=LineItemID(generate_uuid7()),
            article_id=article.id,
            article_number=article.item_number,
            article_type=article.type_,
            name=article.name,
            unit_price=article.price,
            tax_rate=article.tax_rate,
            quantity=quantity,
            line_amount=line_amount,
            processing_required=article.processing_required,
        )


def _build_order_placed_log_entry(
    incoming_order: IncomingOrder,
) -> OrderLogEntry:
    data = {
        'initiator_id': str(incoming_order.orderer.user.id),
    }

    return OrderLogEntry(
        id=generate_uuid7(),
        occurred_at=incoming_order.created_at,
        event_type='order-placed',
        order_id=incoming_order.id,
        data=data,
    )


def add_note(order: Order, author: User, text: str) -> OrderLogEntry:
    log_entry = _build_note_log_entry(order.id, author, text)

    return log_entry


def _build_note_log_entry(
    order_id: OrderID,
    author: User,
    text: str,
) -> OrderLogEntry:
    data = {
        'author_id': str(author.id),
        'text': text,
    }

    return OrderLogEntry(
        id=generate_uuid7(),
        occurred_at=datetime.utcnow(),
        event_type='order-note-added',
        order_id=order_id,
        data=data,
    )


def set_shipped_flag(
    order: Order, initiator: User
) -> Result[OrderLogEntry, str]:
    if not order.is_processing_required:
        return Err('Order contains no items that require shipping.')

    log_entry = _build_set_shipped_flag_log_entry(order.id, initiator)

    return Ok(log_entry)


def _build_set_shipped_flag_log_entry(
    order_id: OrderID, initiator: User
) -> OrderLogEntry:
    data = {
        'initiator_id': str(initiator.id),
    }

    return OrderLogEntry(
        id=generate_uuid7(),
        occurred_at=datetime.utcnow(),
        event_type='order-shipped',
        order_id=order_id,
        data=data,
    )


def unset_shipped_flag(
    order: Order, initiator: User
) -> Result[OrderLogEntry, str]:
    if not order.is_processing_required:
        return Err('Order contains no items that require shipping.')

    log_entry = _build_unset_shipped_flag_log_entry(order.id, initiator)

    return Ok(log_entry)


def _build_unset_shipped_flag_log_entry(
    order_id: OrderID, initiator: User
) -> OrderLogEntry:
    data = {
        'initiator_id': str(initiator.id),
    }

    return OrderLogEntry(
        id=generate_uuid7(),
        occurred_at=datetime.utcnow(),
        event_type='order-shipped-withdrawn',
        order_id=order_id,
        data=data,
    )


def create_payment(
    order: Order,
    created_at: datetime,
    method: str,
    amount: Money,
    initiator: User,
    additional_data: AdditionalPaymentData,
) -> tuple[Payment, OrderLogEntry]:
    payment = _build_payment(
        order.id, created_at, method, amount, additional_data
    )
    log_entry = _build_payment_log_entry(payment, initiator)

    return payment, log_entry


def _build_payment(
    order_id: OrderID,
    created_at: datetime,
    method: str,
    amount: Money,
    additional_data: AdditionalPaymentData,
) -> Payment:
    return Payment(
        id=generate_uuid7(),
        order_id=order_id,
        created_at=created_at,
        method=method,
        amount=amount,
        additional_data=additional_data,
    )


def _build_payment_log_entry(
    payment: Payment, initiator: User
) -> OrderLogEntry:
    data = {
        'payment_id': str(payment.id),
        'initiator_id': str(initiator.id),
    }

    return OrderLogEntry(
        id=generate_uuid7(),
        occurred_at=payment.created_at,
        event_type='order-payment-created',
        order_id=payment.order_id,
        data=data,
    )


def mark_order_as_paid(
    order: Order,
    orderer_user: User,
    occurred_at: datetime,
    payment_method: str,
    additional_payment_data: AdditionalPaymentData | None,
    initiator: User,
) -> Result[
    tuple[ShopOrderPaidEvent, OrderLogEntry],
    OrderAlreadyMarkedAsPaidError,
]:
    if _is_paid(order):
        return Err(OrderAlreadyMarkedAsPaidError())

    payment_state_from = order.payment_state

    event = _build_order_paid_event(
        occurred_at, order, orderer_user, payment_method, initiator
    )

    log_entry = _build_order_paid_log_entry(
        occurred_at,
        order.id,
        payment_state_from,
        payment_method,
        additional_payment_data,
        initiator,
    )

    return Ok((event, log_entry))


def _build_order_paid_event(
    occurred_at: datetime,
    order: Order,
    orderer_user: User,
    payment_method: str,
    initiator: User,
) -> ShopOrderPaidEvent:
    return ShopOrderPaidEvent(
        occurred_at=occurred_at,
        initiator=EventUser.from_user(initiator),
        order_id=order.id,
        order_number=order.order_number,
        orderer=EventUser.from_user(orderer_user),
        payment_method=payment_method,
    )


def _build_order_paid_log_entry(
    occurred_at: datetime,
    order_id: OrderID,
    payment_state_from: PaymentState,
    payment_method: str,
    additional_payment_data: AdditionalPaymentData | None,
    initiator: User,
) -> OrderLogEntry:
    data: OrderLogEntryData = {}

    # Add required, internally set properties after given additional
    # ones to ensure the former are not overridden by the latter.

    if additional_payment_data is not None:
        data.update(additional_payment_data)

    data.update(
        {
            'former_payment_state': payment_state_from.name,
            'payment_method': payment_method,
            'initiator_id': str(initiator.id),
        }
    )

    return OrderLogEntry(
        id=generate_uuid7(),
        occurred_at=occurred_at,
        event_type='order-paid',
        order_id=order_id,
        data=data,
    )


def cancel_order(
    order: Order,
    orderer_user: User,
    occurred_at: datetime,
    reason: str,
    initiator: User,
) -> Result[
    tuple[ShopOrderCanceledEvent, OrderLogEntry],
    OrderAlreadyCanceledError,
]:
    if _is_canceled(order):
        return Err(OrderAlreadyCanceledError())

    has_order_been_paid = _is_paid(order)

    payment_state_from = order.payment_state

    event = _build_order_canceled_event(
        occurred_at, order, orderer_user, initiator
    )

    log_entry = _build_order_canceled_log_entry(
        occurred_at,
        order.id,
        has_order_been_paid,
        payment_state_from,
        reason,
        initiator,
    )

    return Ok((event, log_entry))


def _build_order_canceled_event(
    occurred_at: datetime,
    order: Order,
    orderer_user: User,
    initiator: User,
) -> ShopOrderCanceledEvent:
    return ShopOrderCanceledEvent(
        occurred_at=occurred_at,
        initiator=EventUser.from_user(initiator),
        order_id=order.id,
        order_number=order.order_number,
        orderer=EventUser.from_user(orderer_user),
    )


def _build_order_canceled_log_entry(
    occurred_at: datetime,
    order_id: OrderID,
    has_order_been_paid: bool,
    payment_state_from: PaymentState,
    reason: str,
    initiator: User,
) -> OrderLogEntry:
    event_type = (
        'order-canceled-after-paid'
        if has_order_been_paid
        else 'order-canceled-before-paid'
    )

    data = {
        'former_payment_state': payment_state_from.name,
        'reason': reason,
        'initiator_id': str(initiator.id),
    }

    return OrderLogEntry(
        id=generate_uuid7(),
        occurred_at=occurred_at,
        event_type=event_type,
        order_id=order_id,
        data=data,
    )


def _is_paid(order: Order) -> bool:
    return order.payment_state == PaymentState.paid


def _is_canceled(order: Order) -> bool:
    return order.payment_state in {
        PaymentState.canceled_before_paid,
        PaymentState.canceled_after_paid,
    }


def is_overdue(created_at: datetime, payment_state: PaymentState) -> bool:
    """Return `True` if payment of the order is overdue."""
    if payment_state != PaymentState.open:
        return False

    return datetime.utcnow() >= (created_at + OVERDUE_THRESHOLD)


1		"""
2		byceps.services.shop.order.order_domain_service
3		~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5		:Copyright: 2014-2024 Jochen Kupperschmidt
6		:License: Revised BSD (see `LICENSE` file for details)
7		"""
8
9	1	from collections.abc import Iterator
10		from datetime import datetime, timedelta
11	1
12		from moneyed import Currency, Money
13	1
14		from byceps.events.base import EventUser
15	1	from byceps.events.shop import ShopOrderCanceledEvent, ShopOrderPaidEvent
16	1	from byceps.services.shop.cart.models import Cart, CartItem
17	1	from byceps.services.shop.shop.models import ShopID
18	1	from byceps.services.shop.storefront.models import StorefrontID
19		from byceps.services.user.models.user import User
20	1	from byceps.util.result import Err, Ok, Result
21	1	from byceps.util.uuid import generate_uuid7
22	1
23	1	from .errors import (
24		CartEmpty,
25		OrderAlreadyCanceledError,
26	1	OrderAlreadyMarkedAsPaidError,
27		)
28		from .models.checkout import IncomingLineItem, IncomingOrder
29	1	from .models.log import OrderLogEntry, OrderLogEntryData
30		from .models.order import LineItemID, Order, Orderer, OrderID, PaymentState
31		from .models.payment import AdditionalPaymentData, Payment
32
33
34		OVERDUE_THRESHOLD = timedelta(days=14)
35	1
36
37		def place_order(
38		created_at: datetime,
39		shop_id: ShopID,
40		storefront_id: StorefrontID,
41		orderer: Orderer,
42		currency: Currency,
43		cart: Cart,
44		) -> Result[tuple[IncomingOrder, OrderLogEntry], CartEmpty]:
45		"""Place an order."""
46		cart_items = cart.get_items()
47		if not cart_items:
48		return Err(CartEmpty())
49
50		line_items = list(_build_incoming_line_items(cart_items))
51
52		total_amount = cart.calculate_total_amount()
53
54	1	processing_required = any(
55		line_item.processing_required for line_item in line_items
56		)
57
58		incoming_order = IncomingOrder(
59		id=OrderID(generate_uuid7()),
60		created_at=created_at,
61		shop_id=shop_id,
62		storefront_id=storefront_id,
63		orderer=orderer,
64		line_items=line_items,
65	1	total_amount=total_amount,
66		processing_required=processing_required,
67		)
68
69		log_entry = _build_order_placed_log_entry(incoming_order)
70
71		return Ok((incoming_order, log_entry))
72
73
74		def _build_incoming_line_items(
75		cart_items: list[CartItem],
76		) -> Iterator[IncomingLineItem]:
77		"""Build incoming line item objects from the cart's content."""
78		for cart_item in cart_items:
79		article = cart_item.article
80		quantity = cart_item.quantity
81	1	line_amount = cart_item.line_amount
82
83		yield IncomingLineItem(
84		id=LineItemID(generate_uuid7()),
85		article_id=article.id,
86		article_number=article.item_number,
87		article_type=article.type_,
88		name=article.name,
89		unit_price=article.price,
90		tax_rate=article.tax_rate,
91		quantity=quantity,
92	1	line_amount=line_amount,
93		processing_required=article.processing_required,
94		)
95
96
97		def _build_order_placed_log_entry(
98		incoming_order: IncomingOrder,
99		) -> OrderLogEntry:
100		data = {
101		'initiator_id': str(incoming_order.orderer.user.id),
102		}
103
104		return OrderLogEntry(
105		id=generate_uuid7(),
106		occurred_at=incoming_order.created_at,
107		event_type='order-placed',
108	1	order_id=incoming_order.id,
109		data=data,
110		)
111
112
113		def add_note(order: Order, author: User, text: str) -> OrderLogEntry:
114		log_entry = _build_note_log_entry(order.id, author, text)
115
116	1	return log_entry
117
118
119	1	def _build_note_log_entry(
120		order_id: OrderID,
121	1	author: User,
122		text: str,
123		) -> OrderLogEntry:
124	1	data = {
125		'author_id': str(author.id),
126		'text': text,
127		}
128
129		return OrderLogEntry(
130		id=generate_uuid7(),
131	1	occurred_at=datetime.utcnow(),
132		event_type='order-note-added',
133		order_id=order_id,
134		data=data,
135		)
136
137
138		def set_shipped_flag(
139		order: Order, initiator: User
140		) -> Result[OrderLogEntry, str]:
141	1	if not order.is_processing_required:
142		return Err('Order contains no items that require shipping.')
143
144	1	log_entry = _build_set_shipped_flag_log_entry(order.id, initiator)
145
146		return Ok(log_entry)
147
148
149	1	def _build_set_shipped_flag_log_entry(
150		order_id: OrderID, initiator: User
151		) -> OrderLogEntry:
152		data = {
153		'initiator_id': str(initiator.id),
154		}
155
156		return OrderLogEntry(
157		id=generate_uuid7(),
158	1	occurred_at=datetime.utcnow(),
159		event_type='order-shipped',
160		order_id=order_id,
161		data=data,
162		)
163
164
165		def unset_shipped_flag(
166		order: Order, initiator: User
167		) -> Result[OrderLogEntry, str]:
168		if not order.is_processing_required:
169	1	return Err('Order contains no items that require shipping.')
170
171		log_entry = _build_unset_shipped_flag_log_entry(order.id, initiator)
172	1
173		return Ok(log_entry)
174	1
175
176		def _build_unset_shipped_flag_log_entry(
177		order_id: OrderID, initiator: User
178	1	) -> OrderLogEntry:
179		data = {
180		'initiator_id': str(initiator.id),
181		}
182
183		return OrderLogEntry(
184		id=generate_uuid7(),
185		occurred_at=datetime.utcnow(),
186		event_type='order-shipped-withdrawn',
187	1	order_id=order_id,
188		data=data,
189		)
190	1
191
192		def create_payment(
193		order: Order,
194		created_at: datetime,
195		method: str,
196		amount: Money,
197	1	initiator: User,
198		additional_data: AdditionalPaymentData,
199		) -> tuple[Payment, OrderLogEntry]:
200		payment = _build_payment(
201		order.id, created_at, method, amount, additional_data
202		)
203		log_entry = _build_payment_log_entry(payment, initiator)
204
205		return payment, log_entry
206
207
208		def _build_payment(
209	1	order_id: OrderID,
210		created_at: datetime,
211		method: str,
212		amount: Money,
213		additional_data: AdditionalPaymentData,
214		) -> Payment:
215		return Payment(
216		id=generate_uuid7(),
217	1	order_id=order_id,
218		created_at=created_at,
219		method=method,
220		amount=amount,
221		additional_data=additional_data,
222	1	)
223	1
224
225	1	def _build_payment_log_entry(
226		payment: Payment, initiator: User
227		) -> OrderLogEntry:
228		data = {
229		'payment_id': str(payment.id),
230		'initiator_id': str(initiator.id),
231		}
232
233	1	return OrderLogEntry(
234		id=generate_uuid7(),
235		occurred_at=payment.created_at,
236		event_type='order-payment-created',
237		order_id=payment.order_id,
238		data=data,
239		)
240
241
242	1	def mark_order_as_paid(
243		order: Order,
244		orderer_user: User,
245		occurred_at: datetime,
246		payment_method: str,
247		additional_payment_data: AdditionalPaymentData \| None,
248		initiator: User,
249		) -> Result[
250		tuple[ShopOrderPaidEvent, OrderLogEntry],
251		OrderAlreadyMarkedAsPaidError,
252	1	]:
253		if _is_paid(order):
254		return Err(OrderAlreadyMarkedAsPaidError())
255	1
256		payment_state_from = order.payment_state
257	1
258		event = _build_order_paid_event(
259	1	occurred_at, order, orderer_user, payment_method, initiator
260		)
261
262		log_entry = _build_order_paid_log_entry(
263	1	occurred_at,
264		order.id,
265		payment_state_from,
266		payment_method,
267		additional_payment_data,
268		initiator,
269		)
270
271		return Ok((event, log_entry))
272	1
273
274		def _build_order_paid_event(
275	1	occurred_at: datetime,
276		order: Order,
277		orderer_user: User,
278		payment_method: str,
279		initiator: User,
280		) -> ShopOrderPaidEvent:
281	1	return ShopOrderPaidEvent(
282		occurred_at=occurred_at,
283		initiator=EventUser.from_user(initiator),
284		order_id=order.id,
285		order_number=order.order_number,
286		orderer=EventUser.from_user(orderer_user),
287		payment_method=payment_method,
288		)
289
290
291		def _build_order_paid_log_entry(
292	1	occurred_at: datetime,
293		order_id: OrderID,
294		payment_state_from: PaymentState,
295		payment_method: str,
296		additional_payment_data: AdditionalPaymentData \| None,
297		initiator: User,
298		) -> OrderLogEntry:
299		data: OrderLogEntryData = {}
300	1
301		# Add required, internally set properties after given additional
302		# ones to ensure the former are not overridden by the latter.
303
304		if additional_payment_data is not None:
305		data.update(additional_payment_data)
306	1
307		data.update(
308		{
309		'former_payment_state': payment_state_from.name,
310		'payment_method': payment_method,
311		'initiator_id': str(initiator.id),
312	1	}
313		)
314
315		return OrderLogEntry(
316		id=generate_uuid7(),
317		occurred_at=occurred_at,
318		event_type='order-paid',
319		order_id=order_id,
320		data=data,
321	1	)
322	1
323
324		def cancel_order(
325	1	order: Order,
326	1	orderer_user: User,
327		occurred_at: datetime,
328		reason: str,
329		initiator: User,
330		) -> Result[
331		tuple[ShopOrderCanceledEvent, OrderLogEntry],
332	1	OrderAlreadyCanceledError,
333		]:
334	1	if _is_canceled(order):
335	1	return Err(OrderAlreadyCanceledError())
336
337	1	has_order_been_paid = _is_paid(order)
338
339		payment_state_from = order.payment_state
340
341		event = _build_order_canceled_event(
342		occurred_at, order, orderer_user, initiator
343		)
344
345		log_entry = _build_order_canceled_log_entry(
346		occurred_at,
347		order.id,
348		has_order_been_paid,
349		payment_state_from,
350		reason,
351		initiator,
352		)
353
354		return Ok((event, log_entry))
355
356
357		def _build_order_canceled_event(
358		occurred_at: datetime,
359		order: Order,
360		orderer_user: User,
361		initiator: User,
362		) -> ShopOrderCanceledEvent:
363		return ShopOrderCanceledEvent(
364		occurred_at=occurred_at,
365		initiator=EventUser.from_user(initiator),
366		order_id=order.id,
367		order_number=order.order_number,
368		orderer=EventUser.from_user(orderer_user),
369		)
370
371
372		def _build_order_canceled_log_entry(
373		occurred_at: datetime,
374		order_id: OrderID,
375		has_order_been_paid: bool,
376		payment_state_from: PaymentState,
377		reason: str,
378		initiator: User,
379		) -> OrderLogEntry:
380		event_type = (
381		'order-canceled-after-paid'
382		if has_order_been_paid
383		else 'order-canceled-before-paid'
384		)
385
386		data = {
387		'former_payment_state': payment_state_from.name,
388		'reason': reason,
389		'initiator_id': str(initiator.id),
390		}
391
392		return OrderLogEntry(
393		id=generate_uuid7(),
394		occurred_at=occurred_at,
395		event_type=event_type,
396		order_id=order_id,
397		data=data,
398		)
399
400
401		def _is_paid(order: Order) -> bool:
402		return order.payment_state == PaymentState.paid
403
404
405		def _is_canceled(order: Order) -> bool:
406		return order.payment_state in {
407		PaymentState.canceled_before_paid,
408		PaymentState.canceled_after_paid,
409		}
410
411
412		def is_overdue(created_at: datetime, payment_state: PaymentState) -> bool:
413		"""Return `True` if payment of the order is overdue."""
414		if payment_state != PaymentState.open:
415		return False
416
417		return datetime.utcnow() >= (created_at + OVERDUE_THRESHOLD)
418

byceps / byceps

Push — main ( d0a955...ec822a )

_build_incoming_line_items() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like