giles
278ae3e8f6
SxExpr is now a str subclass so it works everywhere a plain string does (join, isinstance, f-strings) while serialize() still emits it unquoted. sx_call() and all internal render functions (_render_to_sx, async_eval_to_sx, etc.) return SxExpr, eliminating the "forgot to wrap" bug class that caused the sx_content leak and list serialization bugs. - Phase 0: SxExpr(str) with .source property, __add__/__radd__ - Phase 1: sx_call returns SxExpr (drop-in, all 200+ sites unchanged) - Phase 2: async_eval_to_sx, async_eval_slot_to_sx, _render_to_sx, mobile_menu_sx return SxExpr; remove isinstance(str) workaround - Phase 3: Remove ~150 redundant SxExpr() wrappings across 45 files - Phase 4: serialize() docstring, handler return docs, ;; returns: sx Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
114 lines
4.4 KiB
Python
114 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
from quart import Blueprint, g, redirect, url_for, make_response
|
|
from sqlalchemy import select
|
|
|
|
from shared.models.order import Order
|
|
from shared.browser.app.payments.sumup import create_checkout as sumup_create_checkout
|
|
from shared.config import config
|
|
|
|
from shared.infrastructure.cart_identity import current_cart_identity
|
|
from services.check_sumup_status import check_sumup_status
|
|
|
|
from .filters.qs import makeqs_factory, decode
|
|
|
|
|
|
def _owner_filter():
|
|
"""Return SQLAlchemy clause restricting orders to current user/session."""
|
|
ident = current_cart_identity()
|
|
if ident["user_id"]:
|
|
return Order.user_id == ident["user_id"]
|
|
if ident["session_id"]:
|
|
return Order.session_id == ident["session_id"]
|
|
return None
|
|
|
|
|
|
def register() -> Blueprint:
|
|
bp = Blueprint("order", __name__, url_prefix='/<int:order_id>')
|
|
|
|
@bp.before_request
|
|
def route():
|
|
g.makeqs_factory = makeqs_factory
|
|
|
|
@bp.get("/pay/")
|
|
async def order_pay(order_id: int):
|
|
"""Re-open the SumUp payment page for this order."""
|
|
owner = _owner_filter()
|
|
if owner is None:
|
|
return await make_response("Order not found", 404)
|
|
result = await g.s.execute(select(Order).where(Order.id == order_id, owner))
|
|
order = result.scalar_one_or_none()
|
|
if not order:
|
|
return await make_response("Order not found", 404)
|
|
|
|
if order.status == "paid":
|
|
return redirect(url_for("defpage_order_detail", order_id=order.id))
|
|
|
|
if order.sumup_hosted_url:
|
|
return redirect(order.sumup_hosted_url)
|
|
|
|
redirect_url = url_for("checkout.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("checkout.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:
|
|
from shared.sx.page import get_template_context
|
|
from shared.sx.helpers import sx_call, root_header_sx, header_child_sx, full_page_sx, call_url
|
|
from shared.infrastructure.urls import cart_url
|
|
tctx = await get_template_context()
|
|
account_url = call_url(tctx, "account_url", "")
|
|
auth_hdr = sx_call("auth-header-row", account_url=account_url)
|
|
hdr = "(<> " + await root_header_sx(tctx) + " " + await header_child_sx(auth_hdr) + ")"
|
|
filt = sx_call("checkout-error-header")
|
|
order_sx = sx_call("checkout-error-order-id", oid=f"#{order.id}")
|
|
content = sx_call(
|
|
"checkout-error-content",
|
|
msg="No hosted checkout URL returned from SumUp when trying to reopen payment.",
|
|
order=order_sx,
|
|
back_url=cart_url("/"),
|
|
)
|
|
html = await full_page_sx(tctx, header_rows=hdr, filter=filt, content=content)
|
|
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."""
|
|
owner = _owner_filter()
|
|
if owner is None:
|
|
return await make_response("Order not found", 404)
|
|
result = await g.s.execute(select(Order).where(Order.id == order_id, owner))
|
|
order = result.scalar_one_or_none()
|
|
if not order:
|
|
return await make_response("Order not found", 404)
|
|
|
|
if not order.sumup_checkout_id:
|
|
return redirect(url_for("defpage_order_detail", order_id=order.id))
|
|
|
|
try:
|
|
await check_sumup_status(g.s, order)
|
|
except Exception:
|
|
pass
|
|
|
|
return redirect(url_for("defpage_order_detail", order_id=order.id))
|
|
|
|
return bp
|