cancel_overdue_orders.py
"""Cancel open orders older than N days.
:Copyright: 2019-2025 Jan Korneffel, Jochen Kupperschmidt
:License: Revised BSD (see `LICENSE` file for details)
"""
from datetime import timedelta
import click
from flask_babel import force_locale, gettext
from byceps.services.shop.order import (
order_command_service,
order_service,
signals as shop_order_signals,
)
from byceps.services.shop.order.email import order_email_service
from byceps.services.shop.order.errors import (
OrderActionFailedError,
OrderAlreadyCanceledError,
)
from byceps.services.shop.order.models.order import Order
from byceps.util.result import Err, Ok
from _util import call_with_app_context
from _validators import validate_user_screen_name
@click.command()
@click.option('--shop-id', required=True)
@click.option('--minimum-age-in-days', required=True, type=click.INT)
@click.option('--canceler', required=True, callback=validate_user_screen_name)
@click.option('--locale', required=True, default='en')
@click.option('--reason')
@click.option('--notify/--no-notify', required=True)
@click.option('--limit', default=100)
def execute(
shop_id,
minimum_age_in_days: int,
canceler,
locale: str,
reason: str | None,
notify: bool,
limit: int,
):
overdue_orders = _collect_overdue_orders(
shop_id, minimum_age_in_days, limit
)
for order in overdue_orders:
with force_locale(locale):
_cancel_order(order, canceler, reason, notify)
def _collect_overdue_orders(
shop_id, minimum_age_in_days: int, limit: int
) -> list[Order]:
older_than = timedelta(days=minimum_age_in_days)
overdue_orders = order_service.get_overdue_orders(
shop_id, older_than, limit=limit
)
click.secho(f'Found {len(overdue_orders)} overdue orders.', fg='yellow')
return overdue_orders
def _cancel_order(order, canceler, reason: str | None, notify: bool) -> None:
if not reason:
reason = gettext(
'The payment deadline has been exceed. '
'Place a new order if you are still interested in attending.'
)
match order_command_service.cancel_order(order.id, canceler, reason):
case Ok((canceled_order, canceled_event)):
shop_order_signals.order_canceled.send(None, event=canceled_event)
click.secho(
f'Order {order.order_number} was successfully canceled.',
fg='green',
)
if notify:
_notify_orderer(canceled_order)
case Err(e):
if isinstance(e, OrderAlreadyCanceledError):
message = (
f'Order {order.order_number} has already been canceled.'
)
elif isinstance(e, OrderActionFailedError):
message = f'An error occurred on action execution for order {order.order_number}: {e.details}'
else:
message = f'Order {order.order_number} could not be canceled.'
click.secho(message, fg='red')
def _notify_orderer(order) -> None:
order_email_service.send_email_for_canceled_order_to_orderer(order)
click.secho(
f'Notified orderer of canceled order {order.order_number}.', fg='green'
)
if __name__ == '__main__':
call_with_app_context(execute)