Eliminate Python page helpers from orders — pure .sx defpages with IO primitives
Orders defpages now fetch data via (service ...) and generate URLs via (url-for ...) and (route-prefix) directly in .sx. No Python middleman. - Add url-for, route-prefix IO primitives to shared/sx/primitives_io.py - Add generic register()/\_\_getattr\_\_ to ServiceRegistry for dynamic services - Create OrdersPageService with list_page_data/detail_page_data methods - Rewrite orders.sx defpages to use IO primitives + defcomp calls - Remove ~320 lines of Python page helpers from orders/sxc/pages/__init__.py - Convert :data env merge to use kebab-case keys for SX symbol access Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -4,3 +4,6 @@ from __future__ import annotations
|
||||
|
||||
def register_domain_services() -> None:
|
||||
"""Register services for the orders app."""
|
||||
from shared.services.registry import services
|
||||
from .orders_page import OrdersPageService
|
||||
services.register("orders_page", OrdersPageService())
|
||||
|
||||
138
orders/services/orders_page.py
Normal file
138
orders/services/orders_page.py
Normal file
@@ -0,0 +1,138 @@
|
||||
"""Orders page data service — provides serialized dicts for .sx defpages."""
|
||||
from __future__ import annotations
|
||||
|
||||
from sqlalchemy import select, func, or_, cast, String, exists
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from shared.models.order import Order, OrderItem
|
||||
from shared.infrastructure.cart_identity import current_cart_identity
|
||||
from shared.infrastructure.urls import market_product_url
|
||||
|
||||
|
||||
class OrdersPageService:
|
||||
"""Service for orders page data, callable via (service "orders-page" ...)."""
|
||||
|
||||
async def list_page_data(self, session, *, search="", page=1):
|
||||
"""Return orders list + pagination metadata as a dict."""
|
||||
PER_PAGE = 10
|
||||
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:
|
||||
return {"orders": [], "page": 1, "total_pages": 1,
|
||||
"search": "", "search_count": 0}
|
||||
|
||||
page = max(1, int(page))
|
||||
|
||||
where = None
|
||||
if search:
|
||||
term = f"%{search.strip()}%"
|
||||
conds = [
|
||||
Order.status.ilike(term),
|
||||
Order.currency.ilike(term),
|
||||
Order.sumup_checkout_id.ilike(term),
|
||||
Order.sumup_status.ilike(term),
|
||||
Order.description.ilike(term),
|
||||
exists(
|
||||
select(1).select_from(OrderItem)
|
||||
.where(OrderItem.order_id == Order.id,
|
||||
or_(OrderItem.product_title.ilike(term),
|
||||
OrderItem.product_slug.ilike(term)))
|
||||
),
|
||||
]
|
||||
try:
|
||||
conds.append(Order.id == int(search))
|
||||
except (TypeError, ValueError):
|
||||
conds.append(cast(Order.id, String).ilike(term))
|
||||
where = or_(*conds)
|
||||
|
||||
count_q = select(func.count()).select_from(Order).where(owner)
|
||||
if where is not None:
|
||||
count_q = count_q.where(where)
|
||||
total_count = (await session.execute(count_q)).scalar_one() or 0
|
||||
total_pages = max(1, (total_count + PER_PAGE - 1) // PER_PAGE)
|
||||
if page > total_pages:
|
||||
page = total_pages
|
||||
|
||||
stmt = (select(Order).where(owner)
|
||||
.order_by(Order.created_at.desc())
|
||||
.offset((page - 1) * PER_PAGE).limit(PER_PAGE))
|
||||
if where is not None:
|
||||
stmt = stmt.where(where)
|
||||
rows = (await session.execute(stmt)).scalars().all()
|
||||
|
||||
orders = []
|
||||
for o in rows:
|
||||
orders.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}",
|
||||
})
|
||||
|
||||
return {
|
||||
"orders": orders,
|
||||
"page": page,
|
||||
"total_pages": total_pages,
|
||||
"search": search or "",
|
||||
"search_count": total_count,
|
||||
}
|
||||
|
||||
async def detail_page_data(self, session, *, order_id=None):
|
||||
"""Return order detail data as a dict."""
|
||||
from quart import abort
|
||||
|
||||
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 session.execute(
|
||||
select(Order).options(selectinload(Order.items))
|
||||
.where(Order.id == int(order_id), owner)
|
||||
)
|
||||
order = result.scalar_one_or_none()
|
||||
if not order:
|
||||
abort(404)
|
||||
return {}
|
||||
|
||||
items = []
|
||||
for item in (order.items or []):
|
||||
items.append({
|
||||
"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}",
|
||||
})
|
||||
|
||||
return {
|
||||
"order": {
|
||||
"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 "\u2014"),
|
||||
"description": order.description or "",
|
||||
"currency": order.currency or "GBP",
|
||||
"total_formatted": (
|
||||
f"{order.total_amount:.2f}"
|
||||
if order.total_amount else "0.00"),
|
||||
"items": items,
|
||||
},
|
||||
}
|
||||
Reference in New Issue
Block a user