byceps.services.shop.order.service.order_item_to_transfer_object() - Code Metrics - Inspection of "Introduce order state enum" - byceps/byceps - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 854eb5...57d628 )

by Jochen

created 2021-07-19 16:42 UTC

order_item_to_transfer_object() A

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

Complexity

Conditions

Size

Total Lines	12
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	10
nop	1
dl	0
loc	12
ccs	2
cts	2
cp	1
crap	1
rs	9.9
c	0
b	0
f	0

"""
byceps.services.shop.order.service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

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

from __future__ import annotations
from datetime import datetime
from typing import Iterator, Mapping, Optional, Sequence

from flask import current_app
from flask_babel import lazy_gettext
from sqlalchemy.exc import IntegrityError

from ....database import db, paginate, Pagination
from ....events.shop import ShopOrderCanceled, ShopOrderPaid, ShopOrderPlaced
from ....typing import UserID

from ...user import service as user_service

from ..article import service as article_service
from ..cart.models import Cart
from ..shop.dbmodels import Shop as DbShop
from ..shop import service as shop_service
from ..shop.transfer.models import ShopID
from ..storefront import service as storefront_service
from ..storefront.transfer.models import StorefrontID

from .dbmodels.order import Order as DbOrder
from .dbmodels.order_event import OrderEvent as DbOrderEvent, OrderEventData
from .dbmodels.order_item import OrderItem as DbOrderItem
from .models.orderer import Orderer
from . import action_service, event_service, sequence_service
from .transfer.models import (
    Address,
    Order,
    OrderID,
    OrderItem,
    OrderNumber,
    OrderState,
    PaymentMethod,
    PaymentState,
)


class OrderFailed(Exception):
    pass


def place_order(
    storefront_id: StorefrontID,
    orderer: Orderer,
    cart: Cart,
    *,
    created_at: Optional[datetime] = None,
) -> tuple[Order, ShopOrderPlaced]:
    """Place an order for one or more articles."""
    storefront = storefront_service.get_storefront(storefront_id)
    shop = shop_service.get_shop(storefront.shop_id)

    orderer_user = user_service.get_user(orderer.user_id)

    order_number_sequence = sequence_service.get_order_number_sequence(
        storefront.order_number_sequence_id
    )
    order_number = sequence_service.generate_order_number(
        order_number_sequence.id
    )

    order = _build_order(shop.id, order_number, orderer, created_at)
    order_items = list(_build_order_items(cart, order))
    order.total_amount = cart.calculate_total_amount()
    order.shipping_required = any(
        item.shipping_required for item in order_items
    )

    db.session.add(order)
    db.session.add_all(order_items)

    _reduce_article_stock(cart)

    try:
        db.session.commit()
    except IntegrityError as e:
        current_app.logger.error('Order %s failed: %s', order_number, e)
        db.session.rollback()
        raise OrderFailed()

    order_dto = _order_to_transfer_object(order)

    event = ShopOrderPlaced(
        occurred_at=order.created_at,
        initiator_id=orderer_user.id,
        initiator_screen_name=orderer_user.screen_name,
        order_id=order.id,
        order_number=order.order_number,
        orderer_id=orderer_user.id,
        orderer_screen_name=orderer_user.screen_name,
    )

    return order_dto, event


def _build_order(
    shop_id: ShopID,
    order_number: OrderNumber,
    orderer: Orderer,
    created_at: Optional[datetime],
) -> DbOrder:
    """Build an order."""
    return DbOrder(
        shop_id,
        order_number,
        orderer.user_id,
        orderer.first_names,
        orderer.last_name,
        orderer.country,
        orderer.zip_code,
        orderer.city,
        orderer.street,
        created_at=created_at,
    )


def _build_order_items(cart: Cart, order: DbOrder) -> Iterator[DbOrderItem]:
    """Build order items from the cart's content."""
    for cart_item in cart.get_items():
        article = cart_item.article
        quantity = cart_item.quantity
        line_amount = cart_item.line_amount

        yield DbOrderItem(
            order,
            article.item_number,
            article.description,
            article.price,
            article.tax_rate,
            quantity,
            line_amount,
            article.shipping_required,
        )


def _reduce_article_stock(cart: Cart) -> None:
    """Reduce article stock according to what is in the cart."""
    for cart_item in cart.get_items():
        article = cart_item.article
        quantity = cart_item.quantity

        article_service.decrease_quantity(article.id, quantity, commit=False)


def add_note(order_id: OrderID, author_id: UserID, text: str) -> None:
    """Add a note to the order."""
    order = get_order(order_id)
    author = user_service.get_user(author_id)

    event_type = 'order-note-added'
    data = {
        'author_id': str(author.id),
        'text': text,
    }

    event_service.create_event(event_type, order.id, data)


def set_invoiced_flag(order_id: OrderID, initiator_id: UserID) -> None:
    """Record that the invoice for that order has been (externally) created."""
    order = _get_order_entity(order_id)
    initiator = user_service.get_user(initiator_id)

    now = datetime.utcnow()
    event_type = 'order-invoiced'
    data = {
        'initiator_id': str(initiator.id),
    }

    event = DbOrderEvent(now, event_type, order.id, data)
    db.session.add(event)

    order.invoice_created_at = now

    db.session.commit()


def unset_invoiced_flag(order_id: OrderID, initiator_id: UserID) -> None:
    """Withdraw record of the invoice for that order having been created."""
    order = _get_order_entity(order_id)
    initiator = user_service.get_user(initiator_id)

    now = datetime.utcnow()
    event_type = 'order-invoiced-withdrawn'
    data = {
        'initiator_id': str(initiator.id),
    }

    event = DbOrderEvent(now, event_type, order.id, data)
    db.session.add(event)

    order.invoice_created_at = None

    db.session.commit()


def set_shipped_flag(order_id: OrderID, initiator_id: UserID) -> None:
    """Mark the order as shipped."""
    order = _get_order_entity(order_id)
    initiator = user_service.get_user(initiator_id)

    if not order.shipping_required:
        raise ValueError('Order contains no items that require shipping.')

    now = datetime.utcnow()
    event_type = 'order-shipped'
    data = {
        'initiator_id': str(initiator.id),
    }

    event = DbOrderEvent(now, event_type, order.id, data)
    db.session.add(event)

    order.shipped_at = now

    db.session.commit()


def unset_shipped_flag(order_id: OrderID, initiator_id: UserID) -> None:
    """Mark the order as not shipped."""
    order = _get_order_entity(order_id)
    initiator = user_service.get_user(initiator_id)

    if not order.shipping_required:
        raise ValueError('Order contains no items that require shipping.')

    now = datetime.utcnow()
    event_type = 'order-shipped-withdrawn'
    data = {
        'initiator_id': str(initiator.id),
    }

    event = DbOrderEvent(now, event_type, order.id, data)
    db.session.add(event)

    order.shipped_at = None

    db.session.commit()


class OrderAlreadyCanceled(Exception):
    pass


class OrderAlreadyMarkedAsPaid(Exception):
    pass


def cancel_order(
    order_id: OrderID, initiator_id: UserID, reason: str
) -> ShopOrderCanceled:
    """Cancel the order.

    Reserved quantities of articles from that order are made available
    again.
    """
    order = _get_order_entity(order_id)

    if _is_canceled(order):
        raise OrderAlreadyCanceled()

    initiator = user_service.get_user(initiator_id)
    orderer_user = user_service.get_user(order.placed_by_id)

    has_order_been_paid = _is_paid(order)

    now = datetime.utcnow()

    updated_at = now
    payment_state_from = order.payment_state
    payment_state_to = (
        PaymentState.canceled_after_paid
        if has_order_been_paid
        else PaymentState.canceled_before_paid
    )

    _update_payment_state(order, payment_state_to, updated_at, initiator.id)
    order.cancelation_reason = reason

    event_type = (
        'order-canceled-after-paid'
        if has_order_been_paid
        else 'order-canceled-before-paid'
    )
    data = {
        'initiator_id': str(initiator.id),
        'former_payment_state': payment_state_from.name,
        'reason': reason,
    }

    event = DbOrderEvent(now, event_type, order.id, data)
    db.session.add(event)

    # Make the reserved quantity of articles available again.
    for item in order.items:
        article_service.increase_quantity(
            item.article.id, item.quantity, commit=False
        )

    db.session.commit()

    action_service.execute_actions(
        _order_to_transfer_object(order), payment_state_to, initiator.id
    )

    return ShopOrderCanceled(
        occurred_at=updated_at,
        initiator_id=initiator.id,
        initiator_screen_name=initiator.screen_name,
        order_id=order.id,
        order_number=order.order_number,
        orderer_id=orderer_user.id,
        orderer_screen_name=orderer_user.screen_name,
    )


def mark_order_as_paid(
    order_id: OrderID,
    payment_method: PaymentMethod,
    initiator_id: UserID,
    *,
    additional_event_data: Optional[Mapping[str, str]] = None,
) -> ShopOrderPaid:
    """Mark the order as paid."""
    order = _get_order_entity(order_id)

    if _is_paid(order):
        raise OrderAlreadyMarkedAsPaid()

    initiator = user_service.get_user(initiator_id)
    orderer_user = user_service.get_user(order.placed_by_id)

    now = datetime.utcnow()

    updated_at = now
    payment_state_from = order.payment_state
    payment_state_to = PaymentState.paid

    order.payment_method = payment_method
    _update_payment_state(order, payment_state_to, updated_at, initiator.id)

    event_type = 'order-paid'
    # Add required, internally set properties after given additional
    # ones to ensure the former are not overridden by the latter.
    event_data: OrderEventData = {}
    if additional_event_data is not None:
        event_data.update(additional_event_data)
    event_data.update(
        {
            'initiator_id': str(initiator.id),
            'former_payment_state': payment_state_from.name,
            'payment_method': payment_method.name,
        }
    )

    event = DbOrderEvent(now, event_type, order.id, event_data)
    db.session.add(event)

    db.session.commit()

    action_service.execute_actions(
        _order_to_transfer_object(order), payment_state_to, initiator.id
    )

    return ShopOrderPaid(
        occurred_at=updated_at,
        initiator_id=initiator.id,
        initiator_screen_name=initiator.screen_name,
        order_id=order.id,
        order_number=order.order_number,
        orderer_id=orderer_user.id,
        orderer_screen_name=orderer_user.screen_name,
        payment_method=payment_method,
    )


def _update_payment_state(
    order: DbOrder,
    state: PaymentState,
    updated_at: datetime,
    initiator_id: UserID,
) -> None:
    order.payment_state = state
    order.payment_state_updated_at = updated_at
    order.payment_state_updated_by_id = initiator_id


def delete_order(order_id: OrderID) -> None:
    """Delete an order."""
    order = get_order(order_id)

    db.session.query(DbOrderEvent) \
        .filter_by(order_id=order.id) \
        .delete()

    db.session.query(DbOrderItem) \
        .filter_by(order_number=order.order_number) \
        .delete()

    db.session.query(DbOrder) \
        .filter_by(id=order.id) \
        .delete()

    db.session.commit()


def count_open_orders(shop_id: ShopID) -> int:
    """Return the number of open orders for the shop."""
    return DbOrder.query \
        .for_shop(shop_id) \
        .filter_by(_payment_state=PaymentState.open.name) \
        .count()


def count_orders_per_payment_state(shop_id: ShopID) -> dict[PaymentState, int]:
    """Count orders for the shop, grouped by payment state."""
    counts_by_payment_state = dict.fromkeys(PaymentState, 0)

    rows = db.session \
        .query(
            DbOrder._payment_state,
            db.func.count(DbOrder.id)
        ) \
        .filter(DbOrder.shop_id == shop_id) \
        .group_by(DbOrder._payment_state) \
        .all()

    for payment_state_str, count in rows:
        payment_state = PaymentState[payment_state_str]
        counts_by_payment_state[payment_state] = count

    return counts_by_payment_state


def _find_order_entity(order_id: OrderID) -> Optional[DbOrder]:
    """Return the order database entity with that id, or `None` if not
    found.
    """
    return db.session.query(DbOrder).get(order_id)


def _get_order_entity(order_id: OrderID) -> DbOrder:
    """Return the order database entity with that id, or raise an
    exception.
    """
    order = _find_order_entity(order_id)

    if order is None:
        raise ValueError(f'Unknown order ID "{order_id}"')

    return order


def find_order(order_id: OrderID) -> Optional[Order]:
    """Return the order with that id, or `None` if not found."""
    order = _find_order_entity(order_id)

    if order is None:
        return None

    return _order_to_transfer_object(order)


def get_order(order_id: OrderID) -> Order:
    """Return the order with that id, or raise an exception."""
    order = _get_order_entity(order_id)
    return _order_to_transfer_object(order)


def find_order_with_details(order_id: OrderID) -> Optional[Order]:
    """Return the order with that id, or `None` if not found."""
    order = db.session.query(DbOrder) \
        .options(
            db.joinedload(DbOrder.items),
        ) \
        .get(order_id)

    if order is None:
        return None

    return _order_to_transfer_object(order)


def find_order_by_order_number(order_number: OrderNumber) -> Optional[Order]:
    """Return the order with that order number, or `None` if not found."""
    order = DbOrder.query \
        .filter_by(order_number=order_number) \
        .one_or_none()

    if order is None:
        return None

    return _order_to_transfer_object(order)


def find_orders_by_order_numbers(
    order_numbers: set[OrderNumber],
) -> Sequence[Order]:
    """Return the orders with those order numbers."""
    if not order_numbers:
        return []

    orders = DbOrder.query \
        .filter(DbOrder.order_number.in_(order_numbers)) \
        .all()

    return list(map(_order_to_transfer_object, orders))


def get_order_count_by_shop_id() -> dict[ShopID, int]:
    """Return order count (including 0) per shop, indexed by shop ID."""
    shop_ids_and_order_counts = db.session \
        .query(
            DbShop.id,
            db.func.count(DbOrder.shop_id)
        ) \
        .outerjoin(DbOrder) \
        .group_by(DbShop.id) \
        .all()

    return dict(shop_ids_and_order_counts)


def get_orders_for_shop_paginated(
    shop_id: ShopID,
    page: int,
    per_page: int,
    *,
    search_term=None,
    only_payment_state: Optional[PaymentState] = None,
    only_shipped: Optional[bool] = None,
) -> Pagination:
    """Return all orders for that shop, ordered by creation date.

    If a payment state is specified, only orders in that state are
    returned.
    """
    query = DbOrder.query \
        .for_shop(shop_id) \
        .order_by(DbOrder.created_at.desc())

    if search_term:
        ilike_pattern = f'%{search_term}%'
        query = query \
            .filter(DbOrder.order_number.ilike(ilike_pattern))

    if only_payment_state is not None:
        query = query.filter_by(_payment_state=only_payment_state.name)

    if only_shipped is not None:
        query = query.filter(DbOrder.shipping_required == True)

        if only_shipped:
            query = query.filter(DbOrder.shipped_at != None)
        else:
            query = query.filter(DbOrder.shipped_at == None)

    return paginate(
        query,
        page,
        per_page,
        item_mapper=lambda order: _order_to_transfer_object(order),
    )


def get_orders_placed_by_user(user_id: UserID) -> Sequence[Order]:
    """Return orders placed by the user."""
    orders = DbOrder.query \
        .options(
            db.joinedload(DbOrder.items),
        ) \
        .placed_by(user_id) \
        .order_by(DbOrder.created_at.desc()) \
        .all()

    return list(map(_order_to_transfer_object, orders))


def get_orders_placed_by_user_for_shop(
    user_id: UserID, shop_id: ShopID
) -> Sequence[Order]:
    """Return orders placed by the user in that shop."""
    orders = DbOrder.query \
        .options(
            db.joinedload(DbOrder.items),
        ) \
        .for_shop(shop_id) \
        .placed_by(user_id) \
        .order_by(DbOrder.created_at.desc()) \
        .all()

    return list(map(_order_to_transfer_object, orders))


def has_user_placed_orders(user_id: UserID, shop_id: ShopID) -> bool:
    """Return `True` if the user has placed orders in that shop."""
    orders_total = DbOrder.query \
        .for_shop(shop_id) \
        .placed_by(user_id) \
        .count()

    return orders_total > 0


_PAYMENT_METHOD_LABELS = {
    PaymentMethod.bank_transfer: lazy_gettext('bank transfer'),
    PaymentMethod.cash: lazy_gettext('cash'),
    PaymentMethod.direct_debit: lazy_gettext('direct debit'),
    PaymentMethod.free: lazy_gettext('free'),
}


def find_payment_method_label(payment_method: PaymentMethod) -> Optional[str]:
    """Return a label for the payment method."""
    return _PAYMENT_METHOD_LABELS.get(payment_method)


def get_payment_date(order_id: OrderID) -> Optional[datetime]:
    """Return the date the order has been marked as paid, or `None` if
    it has not been paid.
    """
    return db.session \
        .query(DbOrder.payment_state_updated_at) \
        .filter_by(id=order_id) \
        .scalar()


def _order_to_transfer_object(order: DbOrder) -> Order:
    """Create transfer object from order database entity."""
    address = Address(
        country=order.country,
        zip_code=order.zip_code,
        city=order.city,
        street=order.street,
    )

    items = list(map(order_item_to_transfer_object, order.items))

    state = _get_order_state(order)
    is_open = order.payment_state == PaymentState.open
    is_canceled = _is_canceled(order)
    is_paid = _is_paid(order)
    is_invoiced = order.invoice_created_at is not None
    is_shipping_required = order.shipping_required
    is_shipped = order.shipped_at is not None

    return Order(
        id=order.id,
        shop_id=order.shop_id,
        order_number=order.order_number,
        created_at=order.created_at,
        placed_by_id=order.placed_by_id,
        first_names=order.first_names,
        last_name=order.last_name,
        address=address,
        total_amount=order.total_amount,
        items=items,
        payment_method=order.payment_method,
        payment_state=order.payment_state,
        state=state,
        is_open=is_open,
        is_canceled=is_canceled,
        is_paid=is_paid,
        is_invoiced=is_invoiced,
        is_shipping_required=is_shipping_required,
        is_shipped=is_shipped,
        cancelation_reason=order.cancelation_reason,
    )


def order_item_to_transfer_object(
    item: DbOrderItem,
) -> OrderItem:
    """Create transfer object from order item database entity."""
    return OrderItem(
        order_number=item.order_number,
        article_number=item.article_number,
        description=item.description,
        unit_price=item.unit_price,
        tax_rate=item.tax_rate,
        quantity=item.quantity,
        line_amount=item.line_amount,
    )


def _get_order_state(order: DbOrder) -> OrderState:
    is_canceled = _is_canceled(order)
    is_paid = _is_paid(order)
    is_shipping_required = order.shipping_required
    is_shipped = order.shipped_at is not None

    if is_canceled:
        return OrderState.canceled

    if is_paid:
        if not is_shipping_required or is_shipped:
            return OrderState.complete

    return OrderState.open


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


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


1			"""
2			byceps.services.shop.order.service
3			~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5			:Copyright: 2006-2021 Jochen Kupperschmidt
6			:License: Revised BSD (see `LICENSE` file for details)
7			"""
8
9	1		from __future__ import annotations
10	1		from datetime import datetime
11	1		from typing import Iterator, Mapping, Optional, Sequence
12
13	1		from flask import current_app
14	1		from flask_babel import lazy_gettext
15	1		from sqlalchemy.exc import IntegrityError
16
17	1		from ....database import db, paginate, Pagination
18	1		from ....events.shop import ShopOrderCanceled, ShopOrderPaid, ShopOrderPlaced
19	1		from ....typing import UserID
20
21	1		from ...user import service as user_service
22
23	1		from ..article import service as article_service
24	1		from ..cart.models import Cart
25	1		from ..shop.dbmodels import Shop as DbShop
26	1		from ..shop import service as shop_service
27	1		from ..shop.transfer.models import ShopID
28	1		from ..storefront import service as storefront_service
29	1		from ..storefront.transfer.models import StorefrontID
30
31	1		from .dbmodels.order import Order as DbOrder
32	1		from .dbmodels.order_event import OrderEvent as DbOrderEvent, OrderEventData
33	1		from .dbmodels.order_item import OrderItem as DbOrderItem
34	1		from .models.orderer import Orderer
35	1		from . import action_service, event_service, sequence_service
36	1		from .transfer.models import (
37			Address,
38			Order,
39			OrderID,
40			OrderItem,
41			OrderNumber,
42			OrderState,
43			PaymentMethod,
44			PaymentState,
45			)
46
47
48	1		class OrderFailed(Exception):
49	1		pass
50
51
52	1		def place_order(
53			storefront_id: StorefrontID,
54			orderer: Orderer,
55			cart: Cart,
56			*,
57			created_at: Optional[datetime] = None,
58			) -> tuple[Order, ShopOrderPlaced]:
59			"""Place an order for one or more articles."""
60	1		storefront = storefront_service.get_storefront(storefront_id)
61	1		shop = shop_service.get_shop(storefront.shop_id)
62
63	1		orderer_user = user_service.get_user(orderer.user_id)
64
65	1		order_number_sequence = sequence_service.get_order_number_sequence(
66			storefront.order_number_sequence_id
67			)
68	1		order_number = sequence_service.generate_order_number(
69			order_number_sequence.id
70			)
71
72	1		order = _build_order(shop.id, order_number, orderer, created_at)
73	1		order_items = list(_build_order_items(cart, order))
74	1		order.total_amount = cart.calculate_total_amount()
75	1		order.shipping_required = any(
76			item.shipping_required for item in order_items
77			)
78
79	1		db.session.add(order)
80	1		db.session.add_all(order_items)
81
82	1		_reduce_article_stock(cart)
83
84	1		try:
85	1		db.session.commit()
86			except IntegrityError as e:
87			current_app.logger.error('Order %s failed: %s', order_number, e)
88			db.session.rollback()
89			raise OrderFailed()
90
91	1		order_dto = _order_to_transfer_object(order)
92
93	1		event = ShopOrderPlaced(
94			occurred_at=order.created_at,
95			initiator_id=orderer_user.id,
96			initiator_screen_name=orderer_user.screen_name,
97			order_id=order.id,
98			order_number=order.order_number,
99			orderer_id=orderer_user.id,
100			orderer_screen_name=orderer_user.screen_name,
101			)
102
103	1		return order_dto, event
104
105
106	1		def _build_order(
107			shop_id: ShopID,
108			order_number: OrderNumber,
109			orderer: Orderer,
110			created_at: Optional[datetime],
111			) -> DbOrder:
112			"""Build an order."""
113	1		return DbOrder(
114			shop_id,
115			order_number,
116			orderer.user_id,
117			orderer.first_names,
118			orderer.last_name,
119			orderer.country,
120			orderer.zip_code,
121			orderer.city,
122			orderer.street,
123			created_at=created_at,
124			)
125
126
127	1		def _build_order_items(cart: Cart, order: DbOrder) -> Iterator[DbOrderItem]:
128			"""Build order items from the cart's content."""
129	1		for cart_item in cart.get_items():
130	1		article = cart_item.article
131	1		quantity = cart_item.quantity
132	1		line_amount = cart_item.line_amount
133
134	1		yield DbOrderItem(
135			order,
136			article.item_number,
137			article.description,
138			article.price,
139			article.tax_rate,
140			quantity,
141			line_amount,
142			article.shipping_required,
143			)
144
145
146	1		def _reduce_article_stock(cart: Cart) -> None:
147			"""Reduce article stock according to what is in the cart."""
148	1		for cart_item in cart.get_items():
149	1		article = cart_item.article
150	1		quantity = cart_item.quantity
151
152	1		article_service.decrease_quantity(article.id, quantity, commit=False)
153
154
155	1		def add_note(order_id: OrderID, author_id: UserID, text: str) -> None:
156			"""Add a note to the order."""
157			order = get_order(order_id)
158			author = user_service.get_user(author_id)
159
160			event_type = 'order-note-added'
161			data = {
162			'author_id': str(author.id),
163			'text': text,
164			}
165
166			event_service.create_event(event_type, order.id, data)
167
168
169	1		def set_invoiced_flag(order_id: OrderID, initiator_id: UserID) -> None:
170			"""Record that the invoice for that order has been (externally) created."""
171			order = _get_order_entity(order_id)
172			initiator = user_service.get_user(initiator_id)
173
174			now = datetime.utcnow()
175			event_type = 'order-invoiced'
176			data = {
177			'initiator_id': str(initiator.id),
178			}
179
180			event = DbOrderEvent(now, event_type, order.id, data)
181			db.session.add(event)
182
183			order.invoice_created_at = now
184
185			db.session.commit()
186
187
188	1		def unset_invoiced_flag(order_id: OrderID, initiator_id: UserID) -> None:
189			"""Withdraw record of the invoice for that order having been created."""
190			order = _get_order_entity(order_id)
191			initiator = user_service.get_user(initiator_id)
192
193			now = datetime.utcnow()
194			event_type = 'order-invoiced-withdrawn'
195			data = {
196			'initiator_id': str(initiator.id),
197			}
198
199			event = DbOrderEvent(now, event_type, order.id, data)
200			db.session.add(event)
201
202			order.invoice_created_at = None
203
204			db.session.commit()
205
206
207	1	View Code Duplication	def set_shipped_flag(order_id: OrderID, initiator_id: UserID) -> None:
208			"""Mark the order as shipped."""
209			order = _get_order_entity(order_id)
210			initiator = user_service.get_user(initiator_id)
211
212			if not order.shipping_required:
213			raise ValueError('Order contains no items that require shipping.')
214
215			now = datetime.utcnow()
216			event_type = 'order-shipped'
217			data = {
218			'initiator_id': str(initiator.id),
219			}
220
221			event = DbOrderEvent(now, event_type, order.id, data)
222			db.session.add(event)
223
224			order.shipped_at = now
225
226			db.session.commit()
227
228
229	1	View Code Duplication	def unset_shipped_flag(order_id: OrderID, initiator_id: UserID) -> None:
230			"""Mark the order as not shipped."""
231			order = _get_order_entity(order_id)
232			initiator = user_service.get_user(initiator_id)
233
234			if not order.shipping_required:
235			raise ValueError('Order contains no items that require shipping.')
236
237			now = datetime.utcnow()
238			event_type = 'order-shipped-withdrawn'
239			data = {
240			'initiator_id': str(initiator.id),
241			}
242
243			event = DbOrderEvent(now, event_type, order.id, data)
244			db.session.add(event)
245
246			order.shipped_at = None
247
248			db.session.commit()
249
250
251	1		class OrderAlreadyCanceled(Exception):
252	1		pass
253
254
255	1		class OrderAlreadyMarkedAsPaid(Exception):
256	1		pass
257
258
259	1		def cancel_order(
260			order_id: OrderID, initiator_id: UserID, reason: str
261			) -> ShopOrderCanceled:
262			"""Cancel the order.
263
264			Reserved quantities of articles from that order are made available
265			again.
266			"""
267	1		order = _get_order_entity(order_id)
268
269	1		if _is_canceled(order):
270			raise OrderAlreadyCanceled()
271
272	1		initiator = user_service.get_user(initiator_id)
273	1		orderer_user = user_service.get_user(order.placed_by_id)
274
275	1		has_order_been_paid = _is_paid(order)
276
277	1		now = datetime.utcnow()
278
279	1		updated_at = now
280	1		payment_state_from = order.payment_state
281	1		payment_state_to = (
282			PaymentState.canceled_after_paid
283			if has_order_been_paid
284			else PaymentState.canceled_before_paid
285			)
286
287	1		_update_payment_state(order, payment_state_to, updated_at, initiator.id)
288	1		order.cancelation_reason = reason
289
290	1		event_type = (
291			'order-canceled-after-paid'
292			if has_order_been_paid
293			else 'order-canceled-before-paid'
294			)
295	1		data = {
296			'initiator_id': str(initiator.id),
297			'former_payment_state': payment_state_from.name,
298			'reason': reason,
299			}
300
301	1		event = DbOrderEvent(now, event_type, order.id, data)
302	1		db.session.add(event)
303
304			# Make the reserved quantity of articles available again.
305	1		for item in order.items:
306	1		article_service.increase_quantity(
307			item.article.id, item.quantity, commit=False
308			)
309
310	1		db.session.commit()
311
312	1		action_service.execute_actions(
313			_order_to_transfer_object(order), payment_state_to, initiator.id
314			)
315
316	1		return ShopOrderCanceled(
317			occurred_at=updated_at,
318			initiator_id=initiator.id,
319			initiator_screen_name=initiator.screen_name,
320			order_id=order.id,
321			order_number=order.order_number,
322			orderer_id=orderer_user.id,
323			orderer_screen_name=orderer_user.screen_name,
324			)
325
326
327	1		def mark_order_as_paid(
328			order_id: OrderID,
329			payment_method: PaymentMethod,
330			initiator_id: UserID,
331			*,
332			additional_event_data: Optional[Mapping[str, str]] = None,
333			) -> ShopOrderPaid:
334			"""Mark the order as paid."""
335	1		order = _get_order_entity(order_id)
336
337	1		if _is_paid(order):
338			raise OrderAlreadyMarkedAsPaid()
339
340	1		initiator = user_service.get_user(initiator_id)
341	1		orderer_user = user_service.get_user(order.placed_by_id)
342
343	1		now = datetime.utcnow()
344
345	1		updated_at = now
346	1		payment_state_from = order.payment_state
347	1		payment_state_to = PaymentState.paid
348
349	1		order.payment_method = payment_method
350	1		_update_payment_state(order, payment_state_to, updated_at, initiator.id)
351
352	1		event_type = 'order-paid'
353			# Add required, internally set properties after given additional
354			# ones to ensure the former are not overridden by the latter.
355	1		event_data: OrderEventData = {}
356	1		if additional_event_data is not None:
357	1		event_data.update(additional_event_data)
358	1		event_data.update(
359			{
360			'initiator_id': str(initiator.id),
361			'former_payment_state': payment_state_from.name,
362			'payment_method': payment_method.name,
363			}
364			)
365
366	1		event = DbOrderEvent(now, event_type, order.id, event_data)
367	1		db.session.add(event)
368
369	1		db.session.commit()
370
371	1		action_service.execute_actions(
372			_order_to_transfer_object(order), payment_state_to, initiator.id
373			)
374
375	1		return ShopOrderPaid(
376			occurred_at=updated_at,
377			initiator_id=initiator.id,
378			initiator_screen_name=initiator.screen_name,
379			order_id=order.id,
380			order_number=order.order_number,
381			orderer_id=orderer_user.id,
382			orderer_screen_name=orderer_user.screen_name,
383			payment_method=payment_method,
384			)
385
386
387	1		def _update_payment_state(
388			order: DbOrder,
389			state: PaymentState,
390			updated_at: datetime,
391			initiator_id: UserID,
392			) -> None:
393	1		order.payment_state = state
394	1		order.payment_state_updated_at = updated_at
395	1		order.payment_state_updated_by_id = initiator_id
396
397
398	1		def delete_order(order_id: OrderID) -> None:
399			"""Delete an order."""
400	1		order = get_order(order_id)
401
402	1		db.session.query(DbOrderEvent) \
403			.filter_by(order_id=order.id) \
404			.delete()
405
406	1		db.session.query(DbOrderItem) \
407			.filter_by(order_number=order.order_number) \
408			.delete()
409
410	1		db.session.query(DbOrder) \
411			.filter_by(id=order.id) \
412			.delete()
413
414	1		db.session.commit()
415
416
417	1		def count_open_orders(shop_id: ShopID) -> int:
418			"""Return the number of open orders for the shop."""
419			return DbOrder.query \
420			.for_shop(shop_id) \
421			.filter_by(_payment_state=PaymentState.open.name) \
422			.count()
423
424
425	1		def count_orders_per_payment_state(shop_id: ShopID) -> dict[PaymentState, int]:
426			"""Count orders for the shop, grouped by payment state."""
427			counts_by_payment_state = dict.fromkeys(PaymentState, 0)
428
429			rows = db.session \
430			.query(
431			DbOrder._payment_state,
432			db.func.count(DbOrder.id)
433			) \
434			.filter(DbOrder.shop_id == shop_id) \
435			.group_by(DbOrder._payment_state) \
436			.all()
437
438			for payment_state_str, count in rows:
439			payment_state = PaymentState[payment_state_str]
440			counts_by_payment_state[payment_state] = count
441
442			return counts_by_payment_state
443
444
445	1		def _find_order_entity(order_id: OrderID) -> Optional[DbOrder]:
446			"""Return the order database entity with that id, or `None` if not
447			found.
448			"""
449	1		return db.session.query(DbOrder).get(order_id)
450
451
452	1		def _get_order_entity(order_id: OrderID) -> DbOrder:
453			"""Return the order database entity with that id, or raise an
454			exception.
455			"""
456	1		order = _find_order_entity(order_id)
457
458	1		if order is None:
459			raise ValueError(f'Unknown order ID "{order_id}"')
460
461	1		return order
462
463
464	1		def find_order(order_id: OrderID) -> Optional[Order]:
465			"""Return the order with that id, or `None` if not found."""
466	1		order = _find_order_entity(order_id)
467
468	1		if order is None:
469			return None
470
471	1		return _order_to_transfer_object(order)
472
473
474	1		def get_order(order_id: OrderID) -> Order:
475			"""Return the order with that id, or raise an exception."""
476	1		order = _get_order_entity(order_id)
477	1		return _order_to_transfer_object(order)
478
479
480	1		def find_order_with_details(order_id: OrderID) -> Optional[Order]:
481			"""Return the order with that id, or `None` if not found."""
482	1		order = db.session.query(DbOrder) \
483			.options(
484			db.joinedload(DbOrder.items),
485			) \
486			.get(order_id)
487
488	1		if order is None:
489	1		return None
490
491	1		return _order_to_transfer_object(order)
492
493
494	1		def find_order_by_order_number(order_number: OrderNumber) -> Optional[Order]:
495			"""Return the order with that order number, or `None` if not found."""
496			order = DbOrder.query \
497			.filter_by(order_number=order_number) \
498			.one_or_none()
499
500			if order is None:
501			return None
502
503			return _order_to_transfer_object(order)
504
505
506	1		def find_orders_by_order_numbers(
507			order_numbers: set[OrderNumber],
508			) -> Sequence[Order]:
509			"""Return the orders with those order numbers."""
510			if not order_numbers:
511			return []
512
513			orders = DbOrder.query \
514			.filter(DbOrder.order_number.in_(order_numbers)) \
515			.all()
516
517			return list(map(_order_to_transfer_object, orders))
518
519
520	1		def get_order_count_by_shop_id() -> dict[ShopID, int]:
521			"""Return order count (including 0) per shop, indexed by shop ID."""
522			shop_ids_and_order_counts = db.session \
523			.query(
524			DbShop.id,
525			db.func.count(DbOrder.shop_id)
526			) \
527			.outerjoin(DbOrder) \
528			.group_by(DbShop.id) \
529			.all()
530
531			return dict(shop_ids_and_order_counts)
532
533
534	1		def get_orders_for_shop_paginated(
535			shop_id: ShopID,
536			page: int,
537			per_page: int,
538			*,
539			search_term=None,
540			only_payment_state: Optional[PaymentState] = None,
541			only_shipped: Optional[bool] = None,
542			) -> Pagination:
543			"""Return all orders for that shop, ordered by creation date.
544
545			If a payment state is specified, only orders in that state are
546			returned.
547			"""
548			query = DbOrder.query \
549			.for_shop(shop_id) \
550			.order_by(DbOrder.created_at.desc())
551
552			if search_term:
553			ilike_pattern = f'%{search_term}%'
554			query = query \
555			.filter(DbOrder.order_number.ilike(ilike_pattern))
556
557			if only_payment_state is not None:
558			query = query.filter_by(_payment_state=only_payment_state.name)
559
560			if only_shipped is not None:
561			query = query.filter(DbOrder.shipping_required == True)
562
563			if only_shipped:
564			query = query.filter(DbOrder.shipped_at != None)
565			else:
566			query = query.filter(DbOrder.shipped_at == None)
567
568			return paginate(
569			query,
570			page,
571			per_page,
572			item_mapper=lambda order: _order_to_transfer_object(order),
573			)
574
575
576	1		def get_orders_placed_by_user(user_id: UserID) -> Sequence[Order]:
577			"""Return orders placed by the user."""
578	1		orders = DbOrder.query \
579			.options(
580			db.joinedload(DbOrder.items),
581			) \
582			.placed_by(user_id) \
583			.order_by(DbOrder.created_at.desc()) \
584			.all()
585
586	1		return list(map(_order_to_transfer_object, orders))
587
588
589	1		def get_orders_placed_by_user_for_shop(
590			user_id: UserID, shop_id: ShopID
591			) -> Sequence[Order]:
592			"""Return orders placed by the user in that shop."""
593	1		orders = DbOrder.query \
594			.options(
595			db.joinedload(DbOrder.items),
596			) \
597			.for_shop(shop_id) \
598			.placed_by(user_id) \
599			.order_by(DbOrder.created_at.desc()) \
600			.all()
601
602	1		return list(map(_order_to_transfer_object, orders))
603
604
605	1		def has_user_placed_orders(user_id: UserID, shop_id: ShopID) -> bool:
606			"""Return `True` if the user has placed orders in that shop."""
607	1		orders_total = DbOrder.query \
608			.for_shop(shop_id) \
609			.placed_by(user_id) \
610			.count()
611
612	1		return orders_total > 0
613
614
615	1		_PAYMENT_METHOD_LABELS = {
616			PaymentMethod.bank_transfer: lazy_gettext('bank transfer'),
617			PaymentMethod.cash: lazy_gettext('cash'),
618			PaymentMethod.direct_debit: lazy_gettext('direct debit'),
619			PaymentMethod.free: lazy_gettext('free'),
620			}
621
622
623	1		def find_payment_method_label(payment_method: PaymentMethod) -> Optional[str]:
624			"""Return a label for the payment method."""
625	1		return _PAYMENT_METHOD_LABELS.get(payment_method)
626
627
628	1		def get_payment_date(order_id: OrderID) -> Optional[datetime]:
629			"""Return the date the order has been marked as paid, or `None` if
630			it has not been paid.
631			"""
632	1		return db.session \
633			.query(DbOrder.payment_state_updated_at) \
634			.filter_by(id=order_id) \
635			.scalar()
636
637
638	1		def _order_to_transfer_object(order: DbOrder) -> Order:
639			"""Create transfer object from order database entity."""
640	1		address = Address(
641			country=order.country,
642			zip_code=order.zip_code,
643			city=order.city,
644			street=order.street,
645			)
646
647	1		items = list(map(order_item_to_transfer_object, order.items))
648
649	1		state = _get_order_state(order)
650	1		is_open = order.payment_state == PaymentState.open
651	1		is_canceled = _is_canceled(order)
652	1		is_paid = _is_paid(order)
653	1		is_invoiced = order.invoice_created_at is not None
654	1		is_shipping_required = order.shipping_required
655	1		is_shipped = order.shipped_at is not None
656
657	1		return Order(
658			id=order.id,
659			shop_id=order.shop_id,
660			order_number=order.order_number,
661			created_at=order.created_at,
662			placed_by_id=order.placed_by_id,
663			first_names=order.first_names,
664			last_name=order.last_name,
665			address=address,
666			total_amount=order.total_amount,
667			items=items,
668			payment_method=order.payment_method,
669			payment_state=order.payment_state,
670			state=state,
671			is_open=is_open,
672			is_canceled=is_canceled,
673			is_paid=is_paid,
674			is_invoiced=is_invoiced,
675			is_shipping_required=is_shipping_required,
676			is_shipped=is_shipped,
677			cancelation_reason=order.cancelation_reason,
678			)
679
680
681	1		def order_item_to_transfer_object(
682			item: DbOrderItem,
683			) -> OrderItem:
684			"""Create transfer object from order item database entity."""
685	1		return OrderItem(
686			order_number=item.order_number,
687			article_number=item.article_number,
688			description=item.description,
689			unit_price=item.unit_price,
690			tax_rate=item.tax_rate,
691			quantity=item.quantity,
692			line_amount=item.line_amount,
693			)
694
695
696	1		def _get_order_state(order: DbOrder) -> OrderState:
697	1		is_canceled = _is_canceled(order)
698	1		is_paid = _is_paid(order)
699	1		is_shipping_required = order.shipping_required
700	1		is_shipped = order.shipped_at is not None
701
702	1		if is_canceled:
703	1		return OrderState.canceled
704
705	1		if is_paid:
706	1		if not is_shipping_required or is_shipped:
707	1		return OrderState.complete
708
709	1		return OrderState.open
710
711
712	1		def _is_canceled(order: DbOrder) -> bool:
713	1		return order.payment_state in {
714			PaymentState.canceled_before_paid,
715			PaymentState.canceled_after_paid,
716			}
717
718
719	1		def _is_paid(order: DbOrder) -> bool:
720			return order.payment_state == PaymentState.paid
721

byceps / byceps

Push — main ( 854eb5...57d628 )

order_item_to_transfer_object() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like