giles
63b895afd8
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>
139 lines
4.9 KiB
Python
139 lines
4.9 KiB
Python
"""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,
|
|
},
|
|
}
|