967303093d
Some checks failed
Build and Deploy / build-and-deploy (push) Has been cancelled
Extract cart, order, and orders blueprints with their service layer, templates, Dockerfile (APP_MODULE=app:app, IMAGE=cart), entrypoint, and Gitea CI workflow from the coop monolith. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
138 lines
4.9 KiB
Python
138 lines
4.9 KiB
Python
from __future__ import annotations
|
|
|
|
from quart import Blueprint, g, render_template, redirect, url_for, make_response
|
|
from sqlalchemy import select, func, or_, cast, String, exists
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
|
|
from models.market import Product
|
|
from models.order import Order, OrderItem
|
|
from suma_browser.app.payments.sumup import create_checkout as sumup_create_checkout
|
|
from config import config
|
|
|
|
from shared.http_utils import vary as _vary, current_url_without_page as _current_url_without_page
|
|
from suma_browser.app.bp.cart.services import check_sumup_status
|
|
from suma_browser.app.utils.htmx import is_htmx_request
|
|
|
|
from .filters.qs import makeqs_factory, decode
|
|
|
|
|
|
def register() -> Blueprint:
|
|
bp = Blueprint("order", __name__, url_prefix='/<int:order_id>')
|
|
|
|
ORDERS_PER_PAGE = 10 # keep in sync with browse page size / your preference
|
|
|
|
@bp.before_request
|
|
def route():
|
|
# this is the crucial bit for the |qs filter
|
|
g.makeqs_factory = makeqs_factory
|
|
|
|
@bp.get("/")
|
|
async def order_detail(order_id: int):
|
|
"""
|
|
Show a single order + items.
|
|
"""
|
|
result = await g.s.execute(
|
|
select(Order)
|
|
.options(
|
|
selectinload(Order.items).selectinload(OrderItem.product)
|
|
)
|
|
.where(Order.id == order_id)
|
|
)
|
|
order = result.scalar_one_or_none()
|
|
if not order:
|
|
return await make_response("Order not found", 404)
|
|
if not is_htmx_request():
|
|
# Normal browser request: full page with layout
|
|
html = await render_template("_types/order/index.html", order=order,)
|
|
else:
|
|
# HTMX navigation (page 1): main panel + OOB elements
|
|
html = await render_template("_types/order/_oob_elements.html", order=order,)
|
|
|
|
return await make_response(html)
|
|
|
|
@bp.get("/pay/")
|
|
async def order_pay(order_id: int):
|
|
"""
|
|
Re-open the SumUp payment page for this order.
|
|
If already paid, just go back to the order detail.
|
|
If not, (re)create a SumUp checkout and redirect.
|
|
"""
|
|
result = await g.s.execute(select(Order).where(Order.id == order_id))
|
|
order = result.scalar_one_or_none()
|
|
if not order:
|
|
return await make_response("Order not found", 404)
|
|
|
|
if order.status == "paid":
|
|
# Already paid; nothing to pay
|
|
return redirect(url_for("orders.order.order_detail", order_id=order.id))
|
|
|
|
# Prefer to reuse existing hosted URL if we have one
|
|
if order.sumup_hosted_url:
|
|
return redirect(order.sumup_hosted_url)
|
|
|
|
# Otherwise, create a fresh checkout for this order
|
|
redirect_url = url_for("cart.checkout_return", order_id=order.id, _external=True)
|
|
|
|
sumup_cfg = config().get("sumup", {}) or {}
|
|
webhook_secret = sumup_cfg.get("webhook_secret")
|
|
|
|
webhook_url = url_for("cart.checkout_webhook", order_id=order.id, _external=True)
|
|
if webhook_secret:
|
|
from urllib.parse import urlencode
|
|
|
|
sep = "&" if "?" in webhook_url else "?"
|
|
webhook_url = f"{webhook_url}{sep}{urlencode({'token': webhook_secret})}"
|
|
|
|
checkout_data = await sumup_create_checkout(
|
|
order,
|
|
redirect_url=redirect_url,
|
|
webhook_url=webhook_url,
|
|
)
|
|
|
|
order.sumup_checkout_id = checkout_data.get("id")
|
|
order.sumup_status = checkout_data.get("status")
|
|
|
|
hosted_cfg = checkout_data.get("hosted_checkout") or {}
|
|
hosted_url = hosted_cfg.get("hosted_checkout_url") or checkout_data.get("hosted_checkout_url")
|
|
order.sumup_hosted_url = hosted_url
|
|
|
|
await g.s.flush()
|
|
|
|
if not hosted_url:
|
|
html = await render_template(
|
|
"_types/cart/checkout_error.html",
|
|
order=order,
|
|
error="No hosted checkout URL returned from SumUp when trying to reopen payment.",
|
|
)
|
|
return await make_response(html, 500)
|
|
|
|
return redirect(hosted_url)
|
|
|
|
@bp.post("/recheck/")
|
|
async def order_recheck(order_id: int):
|
|
"""
|
|
Manually re-check this order's status with SumUp.
|
|
Useful if the webhook hasn't fired or the user didn't return correctly.
|
|
"""
|
|
result = await g.s.execute(select(Order).where(Order.id == order_id))
|
|
order = result.scalar_one_or_none()
|
|
if not order:
|
|
return await make_response("Order not found", 404)
|
|
|
|
# If we don't have a checkout ID yet, nothing to query
|
|
if not order.sumup_checkout_id:
|
|
return redirect(url_for("orders.order.order_detail", order_id=order.id))
|
|
|
|
try:
|
|
await check_sumup_status(g.s, order)
|
|
except Exception:
|
|
# In a real app, log the error; here we just fall back to previous status
|
|
pass
|
|
|
|
return redirect(url_for("orders.order.order_detail", order_id=order.id))
|
|
|
|
|
|
return bp
|
|
|