byceps.services.shop.order.service.order_item_to_transfer_object() - Code Metrics - Inspection of "Use slightly nicer gray for open order tags" - byceps/byceps - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( d49116...8987e2 )

by Jochen

created 2021-06-17 19:58 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, sequence_service
from .transfer.models import (
    Address,
    Order,
    OrderID,
    OrderItem,
    OrderNumber,
    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 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 order.is_canceled:
        raise OrderAlreadyCanceled()

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

    has_order_been_paid = order.is_paid

    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 order.is_paid:
        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 DbOrder.query.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 = DbOrder.query \
        .options(
            db.joinedload('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('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('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))

    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,
        is_open=order.is_open,
        is_canceled=order.is_canceled,
        is_paid=order.is_paid,
        is_invoiced=order.is_invoiced,
        is_shipping_required=order.is_shipping_required,
        is_shipped=order.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,
    )


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

byceps / byceps

Push — main ( d49116...8987e2 )

order_item_to_transfer_object() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like