"""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, }, }