giles
193578ef88
Eliminate Python s-expression string building across account, orders, federation, and cart services. Visual rendering logic now lives entirely in .sx defcomp components; Python files contain only data serialization, header/layout wiring, and thin wrappers that call defcomps. Phase 0: Shared DRY extraction — auth/orders header defcomps, format-decimal/ pluralize/escape/route-prefix primitives. Phase 1: Account — dashboard, newsletters, login/device/check-email content. Phase 2: Orders — order list, detail, filter, checkout return assembled defcomps. Phase 3: Federation — social nav, post cards, timeline, search, actors, notifications, compose, profile assembled defcomps. Phase 4: Cart — overview, page cart items/calendar/tickets/summary, admin, payments assembled defcomps; orders rendering reuses Phase 2 shared defcomps. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
445 lines
15 KiB
Python
445 lines
15 KiB
Python
"""Orders defpage setup — registers layouts, page helpers, and loads .sx pages."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
|
|
def setup_orders_pages() -> None:
|
|
"""Register orders-specific layouts, page helpers, and load page definitions."""
|
|
_register_orders_layouts()
|
|
_register_orders_helpers()
|
|
_load_orders_page_files()
|
|
|
|
|
|
def _load_orders_page_files() -> None:
|
|
"""Load defpage definitions from orders/sxc/pages/*.sx."""
|
|
import os
|
|
from shared.sx.pages import load_page_dir
|
|
load_page_dir(os.path.dirname(__file__), "orders")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Layouts
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _register_orders_layouts() -> None:
|
|
from shared.sx.layouts import register_custom_layout
|
|
register_custom_layout("orders", _orders_full, _orders_oob, _orders_mobile)
|
|
register_custom_layout("order-detail", _order_detail_full, _order_detail_oob, _order_detail_mobile)
|
|
|
|
|
|
def _orders_full(ctx: dict, **kw: Any) -> str:
|
|
from shared.sx.helpers import root_header_sx, header_child_sx, call_url, sx_call, SxExpr
|
|
|
|
list_url = kw.get("list_url", "/")
|
|
account_url = call_url(ctx, "account_url", "")
|
|
root_hdr = root_header_sx(ctx)
|
|
auth_hdr = sx_call("auth-header-row",
|
|
account_url=account_url,
|
|
select_colours=ctx.get("select_colours", ""),
|
|
account_nav=_as_sx_nav(ctx),
|
|
)
|
|
orders_hdr = sx_call("orders-header-row", list_url=list_url)
|
|
inner = "(<> " + auth_hdr + " " + orders_hdr + ")"
|
|
return "(<> " + root_hdr + " " + header_child_sx(inner) + ")"
|
|
|
|
|
|
def _orders_oob(ctx: dict, **kw: Any) -> str:
|
|
from shared.sx.helpers import root_header_sx, sx_call, SxExpr, call_url
|
|
|
|
list_url = kw.get("list_url", "/")
|
|
account_url = call_url(ctx, "account_url", "")
|
|
auth_hdr = sx_call("auth-header-row",
|
|
account_url=account_url,
|
|
select_colours=ctx.get("select_colours", ""),
|
|
account_nav=_as_sx_nav(ctx),
|
|
oob=True,
|
|
)
|
|
auth_child_oob = sx_call("oob-header-sx",
|
|
parent_id="auth-header-child",
|
|
row=SxExpr(sx_call("orders-header-row", list_url=list_url)))
|
|
root_hdr = root_header_sx(ctx, oob=True)
|
|
return "(<> " + auth_hdr + " " + auth_child_oob + " " + root_hdr + ")"
|
|
|
|
|
|
def _orders_mobile(ctx: dict, **kw: Any) -> str:
|
|
from shared.sx.helpers import mobile_menu_sx, mobile_root_nav_sx
|
|
return mobile_menu_sx(mobile_root_nav_sx(ctx))
|
|
|
|
|
|
def _order_detail_full(ctx: dict, **kw: Any) -> str:
|
|
from shared.sx.helpers import root_header_sx, sx_call, SxExpr, call_url
|
|
|
|
list_url = kw.get("list_url", "/")
|
|
detail_url = kw.get("detail_url", "/")
|
|
account_url = call_url(ctx, "account_url", "")
|
|
root_hdr = root_header_sx(ctx)
|
|
order_row = sx_call(
|
|
"menu-row-sx",
|
|
id="order-row", level=3, colour="sky", link_href=detail_url,
|
|
link_label="Order", icon="fa fa-gbp",
|
|
)
|
|
auth_hdr = sx_call("auth-header-row",
|
|
account_url=account_url,
|
|
select_colours=ctx.get("select_colours", ""),
|
|
account_nav=_as_sx_nav(ctx),
|
|
)
|
|
detail_header = sx_call(
|
|
"order-detail-header-stack",
|
|
auth=SxExpr(auth_hdr),
|
|
orders=SxExpr(sx_call("orders-header-row", list_url=list_url)),
|
|
order=SxExpr(order_row),
|
|
)
|
|
return "(<> " + root_hdr + " " + detail_header + ")"
|
|
|
|
|
|
def _order_detail_oob(ctx: dict, **kw: Any) -> str:
|
|
from shared.sx.helpers import root_header_sx, sx_call, SxExpr
|
|
|
|
detail_url = kw.get("detail_url", "/")
|
|
order_row_oob = sx_call(
|
|
"menu-row-sx",
|
|
id="order-row", level=3, colour="sky", link_href=detail_url,
|
|
link_label="Order", icon="fa fa-gbp", oob=True,
|
|
)
|
|
header_child_oob = sx_call("oob-header-sx",
|
|
parent_id="orders-header-child",
|
|
row=SxExpr(order_row_oob))
|
|
root_hdr = root_header_sx(ctx, oob=True)
|
|
return "(<> " + header_child_oob + " " + root_hdr + ")"
|
|
|
|
|
|
def _order_detail_mobile(ctx: dict, **kw: Any) -> str:
|
|
from shared.sx.helpers import mobile_menu_sx, mobile_root_nav_sx
|
|
return mobile_menu_sx(mobile_root_nav_sx(ctx))
|
|
|
|
|
|
def _as_sx_nav(ctx: dict) -> Any:
|
|
"""Convert account_nav fragment to SxExpr for use in sx_call."""
|
|
from shared.sx.helpers import _as_sx
|
|
return _as_sx(ctx.get("account_nav"))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Page helpers — Python functions callable from defpage expressions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _register_orders_helpers() -> None:
|
|
from shared.sx.pages import register_page_helpers
|
|
|
|
register_page_helpers("orders", {
|
|
# Orders list
|
|
"orders-list-content": _h_orders_list_content,
|
|
"orders-list-filter": _h_orders_list_filter,
|
|
"orders-list-aside": _h_orders_list_aside,
|
|
"orders-list-url": _h_orders_list_url,
|
|
# Order detail
|
|
"order-detail-content": _h_order_detail_content,
|
|
"order-detail-filter": _h_order_detail_filter,
|
|
"order-detail-url": _h_order_detail_url,
|
|
"order-list-url-from-detail": _h_order_list_url_from_detail,
|
|
})
|
|
|
|
|
|
async def _ensure_orders_list():
|
|
"""Fetch orders list data and store in g.orders_page_data."""
|
|
from quart import g, url_for
|
|
if hasattr(g, "orders_page_data"):
|
|
return
|
|
from sqlalchemy import select, func, or_, cast, String, exists
|
|
from shared.models.order import Order, OrderItem
|
|
from shared.infrastructure.cart_identity import current_cart_identity
|
|
from shared.utils import route_prefix
|
|
|
|
ORDERS_PER_PAGE = 10
|
|
ident = current_cart_identity()
|
|
if ident["user_id"]:
|
|
owner_clause = Order.user_id == ident["user_id"]
|
|
elif ident["session_id"]:
|
|
owner_clause = Order.session_id == ident["session_id"]
|
|
else:
|
|
g.orders_page_data = None
|
|
return
|
|
|
|
from bp.orders.filters.qs import makeqs_factory, decode
|
|
q = decode()
|
|
page, search = q.page, q.search
|
|
if page < 1:
|
|
page = 1
|
|
|
|
where_clause = None
|
|
if search:
|
|
term = f"%{search.strip()}%"
|
|
conditions = [
|
|
Order.status.ilike(term),
|
|
Order.currency.ilike(term),
|
|
Order.sumup_checkout_id.ilike(term),
|
|
Order.sumup_status.ilike(term),
|
|
Order.description.ilike(term),
|
|
]
|
|
conditions.append(
|
|
exists(
|
|
select(1).select_from(OrderItem)
|
|
.where(OrderItem.order_id == Order.id,
|
|
or_(OrderItem.product_title.ilike(term),
|
|
OrderItem.product_slug.ilike(term)))
|
|
)
|
|
)
|
|
try:
|
|
search_id = int(search)
|
|
except (TypeError, ValueError):
|
|
search_id = None
|
|
if search_id is not None:
|
|
conditions.append(Order.id == search_id)
|
|
else:
|
|
conditions.append(cast(Order.id, String).ilike(term))
|
|
where_clause = or_(*conditions)
|
|
|
|
count_stmt = select(func.count()).select_from(Order).where(owner_clause)
|
|
if where_clause is not None:
|
|
count_stmt = count_stmt.where(where_clause)
|
|
|
|
total_count_result = await g.s.execute(count_stmt)
|
|
total_count = total_count_result.scalar_one() or 0
|
|
total_pages = max(1, (total_count + ORDERS_PER_PAGE - 1) // ORDERS_PER_PAGE)
|
|
if page > total_pages:
|
|
page = total_pages
|
|
|
|
offset = (page - 1) * ORDERS_PER_PAGE
|
|
stmt = (
|
|
select(Order).where(owner_clause)
|
|
.order_by(Order.created_at.desc())
|
|
.offset(offset).limit(ORDERS_PER_PAGE)
|
|
)
|
|
if where_clause is not None:
|
|
stmt = stmt.where(where_clause)
|
|
|
|
result = await g.s.execute(stmt)
|
|
orders = result.scalars().all()
|
|
pfx = route_prefix()
|
|
qs_fn = makeqs_factory()
|
|
|
|
g.orders_page_data = {
|
|
"orders": orders,
|
|
"page": page,
|
|
"total_pages": total_pages,
|
|
"search": search,
|
|
"search_count": total_count,
|
|
"url_for_fn": url_for,
|
|
"qs_fn": qs_fn,
|
|
"list_url": pfx + url_for("defpage_orders_list"),
|
|
}
|
|
|
|
|
|
async def _ensure_order_detail(order_id):
|
|
"""Fetch order detail data and store in g.order_detail_data."""
|
|
from quart import g, url_for, abort
|
|
if hasattr(g, "order_detail_data"):
|
|
return
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import selectinload
|
|
from shared.models.order import Order
|
|
from shared.infrastructure.cart_identity import current_cart_identity
|
|
from shared.utils import route_prefix
|
|
from shared.browser.app.csrf import generate_csrf_token
|
|
|
|
if order_id is None:
|
|
abort(404)
|
|
|
|
ident = current_cart_identity()
|
|
if ident["user_id"]:
|
|
owner = Order.user_id == ident["user_id"]
|
|
elif ident["session_id"]:
|
|
owner = Order.session_id == ident["session_id"]
|
|
else:
|
|
abort(404)
|
|
return
|
|
|
|
result = await g.s.execute(
|
|
select(Order).options(selectinload(Order.items))
|
|
.where(Order.id == order_id, owner)
|
|
)
|
|
order = result.scalar_one_or_none()
|
|
if not order:
|
|
abort(404)
|
|
return
|
|
|
|
pfx = route_prefix()
|
|
g.order_detail_data = {
|
|
"order": order,
|
|
"calendar_entries": None,
|
|
"detail_url": pfx + url_for("defpage_order_detail", order_id=order.id),
|
|
"list_url": pfx + url_for("defpage_orders_list"),
|
|
"recheck_url": pfx + url_for("orders.order.order_recheck", order_id=order.id),
|
|
"pay_url": pfx + url_for("orders.order.order_pay", order_id=order.id),
|
|
"csrf_token": generate_csrf_token(),
|
|
}
|
|
|
|
|
|
async def _h_orders_list_content(**kw):
|
|
await _ensure_orders_list()
|
|
from quart import g
|
|
from shared.sx.helpers import sx_call, SxExpr
|
|
from shared.sx.parser import serialize
|
|
d = getattr(g, "orders_page_data", None)
|
|
if not d:
|
|
return sx_call("order-empty-state")
|
|
|
|
orders = d["orders"]
|
|
url_for_fn = d["url_for_fn"]
|
|
pfx = d.get("list_url", "/").rsplit("/", 1)[0] if d.get("list_url") else ""
|
|
|
|
order_dicts = []
|
|
for o in orders:
|
|
order_dicts.append({
|
|
"id": o.id,
|
|
"status": o.status or "pending",
|
|
"created_at_formatted": o.created_at.strftime("%-d %b %Y, %H:%M") if o.created_at else "\u2014",
|
|
"description": o.description or "",
|
|
"currency": o.currency or "GBP",
|
|
"total_formatted": f"{o.total_amount or 0:.2f}",
|
|
})
|
|
|
|
from shared.utils import route_prefix
|
|
rpfx = route_prefix()
|
|
detail_prefix = rpfx + url_for_fn("orders.defpage_order_detail", order_id=0).rsplit("0/", 1)[0]
|
|
rows_url = rpfx + url_for_fn("orders.orders_rows")
|
|
|
|
return sx_call("orders-list-content",
|
|
orders=SxExpr(serialize(order_dicts)),
|
|
page=d["page"],
|
|
total_pages=d["total_pages"],
|
|
rows_url=rows_url,
|
|
detail_url_prefix=detail_prefix)
|
|
|
|
|
|
async def _h_orders_list_filter(**kw):
|
|
await _ensure_orders_list()
|
|
from quart import g
|
|
from shared.sx.helpers import sx_call, SxExpr
|
|
from shared.sx.page import SEARCH_HEADERS_MOBILE
|
|
d = getattr(g, "orders_page_data", None)
|
|
search = d.get("search", "") if d else ""
|
|
search_count = d.get("search_count", "") if d else ""
|
|
search_mobile = sx_call("search-mobile",
|
|
current_local_href="/",
|
|
search=search or "",
|
|
search_count=search_count or "",
|
|
hx_select="#main-panel",
|
|
search_headers_mobile=SEARCH_HEADERS_MOBILE,
|
|
)
|
|
return sx_call("order-list-header", search_mobile=SxExpr(search_mobile))
|
|
|
|
|
|
async def _h_orders_list_aside(**kw):
|
|
await _ensure_orders_list()
|
|
from quart import g
|
|
from shared.sx.helpers import sx_call
|
|
from shared.sx.page import SEARCH_HEADERS_DESKTOP
|
|
d = getattr(g, "orders_page_data", None)
|
|
search = d.get("search", "") if d else ""
|
|
search_count = d.get("search_count", "") if d else ""
|
|
return sx_call("search-desktop",
|
|
current_local_href="/",
|
|
search=search or "",
|
|
search_count=search_count or "",
|
|
hx_select="#main-panel",
|
|
search_headers_desktop=SEARCH_HEADERS_DESKTOP,
|
|
)
|
|
|
|
|
|
async def _h_orders_list_url(**kw):
|
|
await _ensure_orders_list()
|
|
from quart import g
|
|
d = getattr(g, "orders_page_data", None)
|
|
return d["list_url"] if d else "/"
|
|
|
|
|
|
async def _h_order_detail_content(order_id=None, **kw):
|
|
await _ensure_order_detail(order_id)
|
|
from quart import g
|
|
from shared.sx.helpers import sx_call, SxExpr
|
|
from shared.sx.parser import serialize
|
|
from shared.infrastructure.urls import market_product_url
|
|
d = getattr(g, "order_detail_data", None)
|
|
if not d:
|
|
return ""
|
|
|
|
order = d["order"]
|
|
order_dict = {
|
|
"id": order.id,
|
|
"status": order.status or "pending",
|
|
"created_at_formatted": order.created_at.strftime("%-d %b %Y, %H:%M") if order.created_at else None,
|
|
"description": order.description,
|
|
"currency": order.currency,
|
|
"total_formatted": f"{order.total_amount:.2f}" if order.total_amount else None,
|
|
"items": [
|
|
{
|
|
"product_url": market_product_url(item.product_slug),
|
|
"product_image": item.product_image,
|
|
"product_title": item.product_title,
|
|
"product_id": item.product_id,
|
|
"quantity": item.quantity,
|
|
"currency": item.currency,
|
|
"unit_price_formatted": f"{item.unit_price or 0:.2f}",
|
|
}
|
|
for item in (order.items or [])
|
|
],
|
|
}
|
|
|
|
cal_entries = d["calendar_entries"]
|
|
cal_dicts = None
|
|
if cal_entries:
|
|
cal_dicts = []
|
|
for e in cal_entries:
|
|
ds = e.start_at.strftime("%-d %b %Y, %H:%M") if e.start_at else ""
|
|
if e.end_at:
|
|
ds += f" \u2013 {e.end_at.strftime('%-d %b %Y, %H:%M')}"
|
|
cal_dicts.append({
|
|
"name": e.name,
|
|
"state": e.state or "",
|
|
"date_str": ds,
|
|
"cost_formatted": f"{e.cost or 0:.2f}",
|
|
})
|
|
|
|
return sx_call("order-detail-content",
|
|
order=SxExpr(serialize(order_dict)),
|
|
calendar_entries=SxExpr(serialize(cal_dicts)) if cal_dicts else None)
|
|
|
|
|
|
async def _h_order_detail_filter(order_id=None, **kw):
|
|
await _ensure_order_detail(order_id)
|
|
from quart import g
|
|
from shared.sx.helpers import sx_call, SxExpr
|
|
from shared.sx.parser import serialize
|
|
d = getattr(g, "order_detail_data", None)
|
|
if not d:
|
|
return ""
|
|
|
|
order = d["order"]
|
|
order_dict = {
|
|
"status": order.status or "pending",
|
|
"created_at_formatted": order.created_at.strftime("%-d %b %Y, %H:%M") if order.created_at else "\u2014",
|
|
}
|
|
|
|
return sx_call("order-detail-filter-content",
|
|
order=SxExpr(serialize(order_dict)),
|
|
list_url=d["list_url"],
|
|
recheck_url=d["recheck_url"],
|
|
pay_url=d["pay_url"],
|
|
csrf=d["csrf_token"])
|
|
|
|
|
|
async def _h_order_detail_url(order_id=None, **kw):
|
|
await _ensure_order_detail(order_id)
|
|
from quart import g
|
|
d = getattr(g, "order_detail_data", None)
|
|
return d["detail_url"] if d else "/"
|
|
|
|
|
|
async def _h_order_list_url_from_detail(order_id=None, **kw):
|
|
await _ensure_order_detail(order_id)
|
|
from quart import g
|
|
d = getattr(g, "order_detail_data", None)
|
|
return d["list_url"] if d else "/"
|