feat: per-page carts with overview, page-scoped checkout, and split blueprints (Phase 4)
Some checks failed
Build and Deploy / build-and-deploy (push) Failing after 21s
Some checks failed
Build and Deploy / build-and-deploy (push) Failing after 21s
Splits the monolithic cart blueprint into three: cart_overview (GET /), page_cart (/<page_slug>/), and cart_global (webhook, return, add). - New page_cart.py service: get_cart_for_page(), get_calendar_entries_for_page(), get_cart_grouped_by_page() - clear_cart_for_order() and create_order_from_cart() accept page_post_id for scoping - Cart app hydrates page_slug via url_value_preprocessor/url_defaults/hydrate_page - Context processor provides page-scoped cart data when g.page_post exists - Internal API /internal/cart/summary accepts ?page_slug= for page-scoped counts - Overview template shows page cards with item counts and totals - Page cart template reuses show_cart() macro with page-specific header Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1,4 +1,6 @@
|
||||
from .cart.routes import register as register_cart_bp
|
||||
from .cart.overview_routes import register as register_cart_overview
|
||||
from .cart.page_routes import register as register_page_cart
|
||||
from .cart.global_routes import register as register_cart_global
|
||||
from .cart.api import register as register_cart_api
|
||||
from .order.routes import register as register_order
|
||||
from .orders.routes import register as register_orders
|
||||
|
||||
@@ -11,7 +11,9 @@ from sqlalchemy import select, update, func
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from models.market import CartItem
|
||||
from models.calendars import CalendarEntry
|
||||
from models.market_place import MarketPlace
|
||||
from models.calendars import CalendarEntry, Calendar
|
||||
from models.ghost_content import Post
|
||||
from suma_browser.app.csrf import csrf_exempt
|
||||
from shared.cart_identity import current_cart_identity
|
||||
|
||||
@@ -26,22 +28,41 @@ def register() -> Blueprint:
|
||||
Return a lightweight cart summary (count + total) for the
|
||||
current session/user. Called by coop and market apps to
|
||||
populate the cart-mini widget without importing cart services.
|
||||
|
||||
Optional query param: ?page_slug=<slug>
|
||||
When provided, returns only items scoped to that page.
|
||||
"""
|
||||
ident = current_cart_identity()
|
||||
|
||||
# --- product cart ---
|
||||
cart_filters = [CartItem.deleted_at.is_(None)]
|
||||
if ident["user_id"] is not None:
|
||||
cart_filters.append(CartItem.user_id == ident["user_id"])
|
||||
else:
|
||||
cart_filters.append(CartItem.session_id == ident["session_id"])
|
||||
# Resolve optional page filter
|
||||
page_slug = request.args.get("page_slug")
|
||||
page_post_id = None
|
||||
if page_slug:
|
||||
post = (
|
||||
await g.s.execute(
|
||||
select(Post).where(Post.slug == page_slug, Post.is_page == True) # noqa: E712
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
if post:
|
||||
page_post_id = post.id
|
||||
|
||||
result = await g.s.execute(
|
||||
select(CartItem)
|
||||
.where(*cart_filters)
|
||||
.options(selectinload(CartItem.product))
|
||||
.order_by(CartItem.created_at.desc())
|
||||
)
|
||||
# --- product cart ---
|
||||
cart_q = select(CartItem).where(CartItem.deleted_at.is_(None))
|
||||
if ident["user_id"] is not None:
|
||||
cart_q = cart_q.where(CartItem.user_id == ident["user_id"])
|
||||
else:
|
||||
cart_q = cart_q.where(CartItem.session_id == ident["session_id"])
|
||||
|
||||
if page_post_id is not None:
|
||||
mp_ids = select(MarketPlace.id).where(
|
||||
MarketPlace.post_id == page_post_id,
|
||||
MarketPlace.deleted_at.is_(None),
|
||||
).scalar_subquery()
|
||||
cart_q = cart_q.where(CartItem.market_place_id.in_(mp_ids))
|
||||
|
||||
cart_q = cart_q.options(selectinload(CartItem.product)).order_by(CartItem.created_at.desc())
|
||||
|
||||
result = await g.s.execute(cart_q)
|
||||
cart_items = result.scalars().all()
|
||||
|
||||
cart_count = sum(ci.quantity for ci in cart_items)
|
||||
@@ -52,18 +73,23 @@ def register() -> Blueprint:
|
||||
)
|
||||
|
||||
# --- calendar entries ---
|
||||
cal_filters = [
|
||||
cal_q = select(CalendarEntry).where(
|
||||
CalendarEntry.deleted_at.is_(None),
|
||||
CalendarEntry.state == "pending",
|
||||
]
|
||||
if ident["user_id"] is not None:
|
||||
cal_filters.append(CalendarEntry.user_id == ident["user_id"])
|
||||
else:
|
||||
cal_filters.append(CalendarEntry.session_id == ident["session_id"])
|
||||
|
||||
cal_result = await g.s.execute(
|
||||
select(CalendarEntry).where(*cal_filters)
|
||||
)
|
||||
if ident["user_id"] is not None:
|
||||
cal_q = cal_q.where(CalendarEntry.user_id == ident["user_id"])
|
||||
else:
|
||||
cal_q = cal_q.where(CalendarEntry.session_id == ident["session_id"])
|
||||
|
||||
if page_post_id is not None:
|
||||
cal_ids = select(Calendar.id).where(
|
||||
Calendar.post_id == page_post_id,
|
||||
Calendar.deleted_at.is_(None),
|
||||
).scalar_subquery()
|
||||
cal_q = cal_q.where(CalendarEntry.calendar_id.in_(cal_ids))
|
||||
|
||||
cal_result = await g.s.execute(cal_q)
|
||||
cal_entries = cal_result.scalars().all()
|
||||
|
||||
calendar_count = len(cal_entries)
|
||||
|
||||
201
bp/cart/global_routes.py
Normal file
201
bp/cart/global_routes.py
Normal file
@@ -0,0 +1,201 @@
|
||||
# bp/cart/global_routes.py — Global cart routes (webhook, return, add)
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from quart import Blueprint, g, request, render_template, redirect, url_for, make_response
|
||||
from sqlalchemy import select
|
||||
|
||||
from models.order import Order
|
||||
from suma_browser.app.utils.htmx import is_htmx_request
|
||||
from .services import (
|
||||
current_cart_identity,
|
||||
get_cart,
|
||||
total,
|
||||
clear_cart_for_order,
|
||||
get_calendar_cart_entries,
|
||||
calendar_total,
|
||||
check_sumup_status,
|
||||
)
|
||||
from .services.checkout import (
|
||||
find_or_create_cart_item,
|
||||
create_order_from_cart,
|
||||
resolve_page_config,
|
||||
build_sumup_description,
|
||||
build_sumup_reference,
|
||||
build_webhook_url,
|
||||
validate_webhook_secret,
|
||||
get_order_with_details,
|
||||
)
|
||||
from suma_browser.app.payments.sumup import create_checkout as sumup_create_checkout
|
||||
from config import config
|
||||
|
||||
|
||||
def register(url_prefix: str) -> Blueprint:
|
||||
bp = Blueprint("cart_global", __name__, url_prefix=url_prefix)
|
||||
|
||||
@bp.post("/add/<int:product_id>/")
|
||||
async def add_to_cart(product_id: int):
|
||||
ident = current_cart_identity()
|
||||
|
||||
cart_item = await find_or_create_cart_item(
|
||||
g.s,
|
||||
product_id,
|
||||
ident["user_id"],
|
||||
ident["session_id"],
|
||||
)
|
||||
|
||||
if not cart_item:
|
||||
return await make_response("Product not found", 404)
|
||||
|
||||
if request.headers.get("HX-Request") == "true":
|
||||
# Redirect to overview for HTMX
|
||||
return redirect(url_for("cart_overview.overview"))
|
||||
|
||||
return redirect(url_for("cart_overview.overview"))
|
||||
|
||||
@bp.post("/checkout/")
|
||||
async def checkout():
|
||||
"""Legacy global checkout (for orphan items without page scope)."""
|
||||
cart = await get_cart(g.s)
|
||||
calendar_entries = await get_calendar_cart_entries(g.s)
|
||||
|
||||
if not cart and not calendar_entries:
|
||||
return redirect(url_for("cart_overview.overview"))
|
||||
|
||||
product_total = total(cart) or 0
|
||||
calendar_amount = calendar_total(calendar_entries) or 0
|
||||
cart_total = product_total + calendar_amount
|
||||
|
||||
if cart_total <= 0:
|
||||
return redirect(url_for("cart_overview.overview"))
|
||||
|
||||
try:
|
||||
page_config = await resolve_page_config(g.s, cart, calendar_entries)
|
||||
except ValueError as e:
|
||||
html = await render_template(
|
||||
"_types/cart/checkout_error.html",
|
||||
order=None,
|
||||
error=str(e),
|
||||
)
|
||||
return await make_response(html, 400)
|
||||
|
||||
ident = current_cart_identity()
|
||||
order = await create_order_from_cart(
|
||||
g.s,
|
||||
cart,
|
||||
calendar_entries,
|
||||
ident.get("user_id"),
|
||||
ident.get("session_id"),
|
||||
product_total,
|
||||
calendar_amount,
|
||||
)
|
||||
|
||||
if page_config:
|
||||
order.page_config_id = page_config.id
|
||||
|
||||
redirect_url = url_for("cart_global.checkout_return", order_id=order.id, _external=True)
|
||||
order.sumup_reference = build_sumup_reference(order.id, page_config=page_config)
|
||||
description = build_sumup_description(cart, order.id)
|
||||
|
||||
webhook_base_url = url_for("cart_global.checkout_webhook", order_id=order.id, _external=True)
|
||||
webhook_url = build_webhook_url(webhook_base_url)
|
||||
|
||||
checkout_data = await sumup_create_checkout(
|
||||
order,
|
||||
redirect_url=redirect_url,
|
||||
webhook_url=webhook_url,
|
||||
description=description,
|
||||
page_config=page_config,
|
||||
)
|
||||
await clear_cart_for_order(g.s, order)
|
||||
|
||||
order.sumup_checkout_id = checkout_data.get("id")
|
||||
order.sumup_status = checkout_data.get("status")
|
||||
order.description = checkout_data.get("description")
|
||||
|
||||
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:
|
||||
html = await render_template(
|
||||
"_types/cart/checkout_error.html",
|
||||
order=order,
|
||||
error="No hosted checkout URL returned from SumUp.",
|
||||
)
|
||||
return await make_response(html, 500)
|
||||
|
||||
return redirect(hosted_url)
|
||||
|
||||
@bp.post("/checkout/webhook/<int:order_id>/")
|
||||
async def checkout_webhook(order_id: int):
|
||||
"""Webhook endpoint for SumUp CHECKOUT_STATUS_CHANGED events."""
|
||||
if not validate_webhook_secret(request.args.get("token")):
|
||||
return "", 204
|
||||
|
||||
try:
|
||||
payload = await request.get_json()
|
||||
except Exception:
|
||||
payload = None
|
||||
|
||||
if not isinstance(payload, dict):
|
||||
return "", 204
|
||||
|
||||
if payload.get("event_type") != "CHECKOUT_STATUS_CHANGED":
|
||||
return "", 204
|
||||
|
||||
checkout_id = payload.get("id")
|
||||
if not checkout_id:
|
||||
return "", 204
|
||||
|
||||
result = await g.s.execute(select(Order).where(Order.id == order_id))
|
||||
order = result.scalar_one_or_none()
|
||||
if not order:
|
||||
return "", 204
|
||||
|
||||
if order.sumup_checkout_id and order.sumup_checkout_id != checkout_id:
|
||||
return "", 204
|
||||
|
||||
try:
|
||||
await check_sumup_status(g.s, order)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return "", 204
|
||||
|
||||
@bp.get("/checkout/return/<int:order_id>/")
|
||||
async def checkout_return(order_id: int):
|
||||
"""Handle the browser returning from SumUp after payment."""
|
||||
order = await get_order_with_details(g.s, order_id)
|
||||
|
||||
if not order:
|
||||
html = await render_template(
|
||||
"_types/cart/checkout_return.html",
|
||||
order=None,
|
||||
status="missing",
|
||||
calendar_entries=[],
|
||||
)
|
||||
return await make_response(html)
|
||||
|
||||
status = (order.status or "pending").lower()
|
||||
|
||||
if order.sumup_checkout_id:
|
||||
try:
|
||||
await check_sumup_status(g.s, order)
|
||||
except Exception:
|
||||
status = status or "pending"
|
||||
|
||||
calendar_entries = order.calendar_entries or []
|
||||
await g.s.flush()
|
||||
|
||||
html = await render_template(
|
||||
"_types/cart/checkout_return.html",
|
||||
order=order,
|
||||
status=status,
|
||||
calendar_entries=calendar_entries,
|
||||
)
|
||||
return await make_response(html)
|
||||
|
||||
return bp
|
||||
31
bp/cart/overview_routes.py
Normal file
31
bp/cart/overview_routes.py
Normal file
@@ -0,0 +1,31 @@
|
||||
# bp/cart/overview_routes.py — Cart overview (list of page carts)
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from quart import Blueprint, render_template, make_response
|
||||
|
||||
from suma_browser.app.utils.htmx import is_htmx_request
|
||||
from .services import get_cart_grouped_by_page
|
||||
|
||||
|
||||
def register(url_prefix: str) -> Blueprint:
|
||||
bp = Blueprint("cart_overview", __name__, url_prefix=url_prefix)
|
||||
|
||||
@bp.get("/")
|
||||
async def overview():
|
||||
from quart import g
|
||||
page_groups = await get_cart_grouped_by_page(g.s)
|
||||
|
||||
if not is_htmx_request():
|
||||
html = await render_template(
|
||||
"_types/cart/overview/index.html",
|
||||
page_groups=page_groups,
|
||||
)
|
||||
else:
|
||||
html = await render_template(
|
||||
"_types/cart/overview/_oob_elements.html",
|
||||
page_groups=page_groups,
|
||||
)
|
||||
return await make_response(html)
|
||||
|
||||
return bp
|
||||
123
bp/cart/page_routes.py
Normal file
123
bp/cart/page_routes.py
Normal file
@@ -0,0 +1,123 @@
|
||||
# bp/cart/page_routes.py — Per-page cart (view + checkout)
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from quart import Blueprint, g, render_template, redirect, make_response, url_for
|
||||
|
||||
from suma_browser.app.utils.htmx import is_htmx_request
|
||||
from suma_browser.app.payments.sumup import create_checkout as sumup_create_checkout
|
||||
from config import config
|
||||
from .services import (
|
||||
total,
|
||||
clear_cart_for_order,
|
||||
calendar_total,
|
||||
check_sumup_status,
|
||||
)
|
||||
from .services.page_cart import get_cart_for_page, get_calendar_entries_for_page
|
||||
from .services.checkout import (
|
||||
create_order_from_cart,
|
||||
build_sumup_description,
|
||||
build_sumup_reference,
|
||||
build_webhook_url,
|
||||
get_order_with_details,
|
||||
)
|
||||
from .services import current_cart_identity
|
||||
|
||||
|
||||
def register(url_prefix: str) -> Blueprint:
|
||||
bp = Blueprint("page_cart", __name__, url_prefix=url_prefix)
|
||||
|
||||
@bp.get("/")
|
||||
async def page_view():
|
||||
post = g.page_post
|
||||
cart = await get_cart_for_page(g.s, post.id)
|
||||
cal_entries = await get_calendar_entries_for_page(g.s, post.id)
|
||||
|
||||
tpl_ctx = dict(
|
||||
page_post=post,
|
||||
page_config=getattr(g, "page_config", None),
|
||||
cart=cart,
|
||||
calendar_cart_entries=cal_entries,
|
||||
total=total,
|
||||
calendar_total=calendar_total,
|
||||
)
|
||||
|
||||
if not is_htmx_request():
|
||||
html = await render_template("_types/cart/page/index.html", **tpl_ctx)
|
||||
else:
|
||||
html = await render_template("_types/cart/page/_oob_elements.html", **tpl_ctx)
|
||||
return await make_response(html)
|
||||
|
||||
@bp.post("/checkout/")
|
||||
async def page_checkout():
|
||||
post = g.page_post
|
||||
page_config = getattr(g, "page_config", None)
|
||||
|
||||
cart = await get_cart_for_page(g.s, post.id)
|
||||
cal_entries = await get_calendar_entries_for_page(g.s, post.id)
|
||||
|
||||
if not cart and not cal_entries:
|
||||
return redirect(url_for("page_cart.page_view"))
|
||||
|
||||
product_total = total(cart) or 0
|
||||
calendar_amount = calendar_total(cal_entries) or 0
|
||||
cart_total = product_total + calendar_amount
|
||||
|
||||
if cart_total <= 0:
|
||||
return redirect(url_for("page_cart.page_view"))
|
||||
|
||||
# Create order scoped to this page
|
||||
ident = current_cart_identity()
|
||||
order = await create_order_from_cart(
|
||||
g.s,
|
||||
cart,
|
||||
cal_entries,
|
||||
ident.get("user_id"),
|
||||
ident.get("session_id"),
|
||||
product_total,
|
||||
calendar_amount,
|
||||
page_post_id=post.id,
|
||||
)
|
||||
|
||||
# Set page_config on order
|
||||
if page_config:
|
||||
order.page_config_id = page_config.id
|
||||
|
||||
# Build SumUp checkout details — webhook/return use global routes
|
||||
redirect_url = url_for("cart_global.checkout_return", order_id=order.id, _external=True)
|
||||
order.sumup_reference = build_sumup_reference(order.id, page_config=page_config)
|
||||
description = build_sumup_description(cart, order.id)
|
||||
|
||||
webhook_base_url = url_for("cart_global.checkout_webhook", order_id=order.id, _external=True)
|
||||
webhook_url = build_webhook_url(webhook_base_url)
|
||||
|
||||
checkout_data = await sumup_create_checkout(
|
||||
order,
|
||||
redirect_url=redirect_url,
|
||||
webhook_url=webhook_url,
|
||||
description=description,
|
||||
page_config=page_config,
|
||||
)
|
||||
await clear_cart_for_order(g.s, order, page_post_id=post.id)
|
||||
|
||||
order.sumup_checkout_id = checkout_data.get("id")
|
||||
order.sumup_status = checkout_data.get("status")
|
||||
order.description = checkout_data.get("description")
|
||||
|
||||
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:
|
||||
html = await render_template(
|
||||
"_types/cart/checkout_error.html",
|
||||
order=order,
|
||||
error="No hosted checkout URL returned from SumUp.",
|
||||
)
|
||||
return await make_response(html, 500)
|
||||
|
||||
return redirect(hosted_url)
|
||||
|
||||
return bp
|
||||
@@ -5,4 +5,9 @@ from .clear_cart_for_order import clear_cart_for_order
|
||||
from .adopt_session_cart_for_user import adopt_session_cart_for_user
|
||||
from .calendar_cart import get_calendar_cart_entries, calendar_total
|
||||
from .check_sumup_status import check_sumup_status
|
||||
from .page_cart import (
|
||||
get_cart_for_page,
|
||||
get_calendar_entries_for_page,
|
||||
get_cart_grouped_by_page,
|
||||
)
|
||||
|
||||
|
||||
@@ -105,10 +105,15 @@ async def create_order_from_cart(
|
||||
session_id: Optional[str],
|
||||
product_total: float,
|
||||
calendar_total: float,
|
||||
*,
|
||||
page_post_id: int | None = None,
|
||||
) -> Order:
|
||||
"""
|
||||
Create an Order and OrderItems from the current cart + calendar entries.
|
||||
Returns the created Order.
|
||||
|
||||
When *page_post_id* is given, only calendar entries whose calendar
|
||||
belongs to that page are marked as "ordered". Otherwise all pending
|
||||
entries are updated (legacy behaviour).
|
||||
"""
|
||||
cart_total = product_total + calendar_total
|
||||
|
||||
@@ -151,6 +156,13 @@ async def create_order_from_cart(
|
||||
elif order.session_id is not None:
|
||||
calendar_filters.append(CalendarEntry.session_id == order.session_id)
|
||||
|
||||
if page_post_id is not None:
|
||||
cal_ids = select(Calendar.id).where(
|
||||
Calendar.post_id == page_post_id,
|
||||
Calendar.deleted_at.is_(None),
|
||||
).scalar_subquery()
|
||||
calendar_filters.append(CalendarEntry.calendar_id.in_(cal_ids))
|
||||
|
||||
await session.execute(
|
||||
update(CalendarEntry)
|
||||
.where(*calendar_filters)
|
||||
|
||||
@@ -1,14 +1,16 @@
|
||||
from sqlalchemy import update, func
|
||||
from sqlalchemy import update, func, select
|
||||
|
||||
from models.market import CartItem
|
||||
from models.market_place import MarketPlace
|
||||
from models.order import Order
|
||||
# ...
|
||||
|
||||
# helper function near the top of the file (outside register())
|
||||
async def clear_cart_for_order(session, order: Order) -> None:
|
||||
|
||||
async def clear_cart_for_order(session, order: Order, *, page_post_id: int | None = None) -> None:
|
||||
"""
|
||||
Soft-delete all CartItem rows belonging to this order's user_id/session_id.
|
||||
Called when an order is marked as paid.
|
||||
Soft-delete CartItem rows belonging to this order's user_id/session_id.
|
||||
|
||||
When *page_post_id* is given, only items whose market_place belongs to
|
||||
that page are cleared. Otherwise all items are cleared (legacy behaviour).
|
||||
"""
|
||||
filters = [CartItem.deleted_at.is_(None)]
|
||||
if order.user_id is not None:
|
||||
@@ -20,6 +22,13 @@ async def clear_cart_for_order(session, order: Order) -> None:
|
||||
# no user_id/session_id on order – nothing to clear
|
||||
return
|
||||
|
||||
if page_post_id is not None:
|
||||
mp_ids = select(MarketPlace.id).where(
|
||||
MarketPlace.post_id == page_post_id,
|
||||
MarketPlace.deleted_at.is_(None),
|
||||
).scalar_subquery()
|
||||
filters.append(CartItem.market_place_id.in_(mp_ids))
|
||||
|
||||
await session.execute(
|
||||
update(CartItem)
|
||||
.where(*filters)
|
||||
|
||||
@@ -18,7 +18,8 @@ async def get_cart(session):
|
||||
.where(*filters)
|
||||
.order_by(CartItem.created_at.desc())
|
||||
.options(
|
||||
selectinload(CartItem.product), # <-- important bit
|
||||
selectinload(CartItem.product),
|
||||
selectinload(CartItem.market_place),
|
||||
)
|
||||
)
|
||||
return result.scalars().all()
|
||||
|
||||
164
bp/cart/services/page_cart.py
Normal file
164
bp/cart/services/page_cart.py
Normal file
@@ -0,0 +1,164 @@
|
||||
"""
|
||||
Page-scoped cart queries.
|
||||
|
||||
Groups cart items and calendar entries by their owning page (Post),
|
||||
determined via CartItem.market_place.post_id and CalendarEntry.calendar.post_id.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import defaultdict
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from models.market import CartItem
|
||||
from models.market_place import MarketPlace
|
||||
from models.calendars import CalendarEntry, Calendar
|
||||
from models.ghost_content import Post
|
||||
from models.page_config import PageConfig
|
||||
from .identity import current_cart_identity
|
||||
|
||||
|
||||
async def get_cart_for_page(session, post_id: int) -> list[CartItem]:
|
||||
"""Return cart items scoped to a specific page (via MarketPlace.post_id)."""
|
||||
ident = current_cart_identity()
|
||||
|
||||
filters = [
|
||||
CartItem.deleted_at.is_(None),
|
||||
MarketPlace.post_id == post_id,
|
||||
MarketPlace.deleted_at.is_(None),
|
||||
]
|
||||
if ident["user_id"] is not None:
|
||||
filters.append(CartItem.user_id == ident["user_id"])
|
||||
else:
|
||||
filters.append(CartItem.session_id == ident["session_id"])
|
||||
|
||||
result = await session.execute(
|
||||
select(CartItem)
|
||||
.join(MarketPlace, CartItem.market_place_id == MarketPlace.id)
|
||||
.where(*filters)
|
||||
.order_by(CartItem.created_at.desc())
|
||||
.options(
|
||||
selectinload(CartItem.product),
|
||||
selectinload(CartItem.market_place),
|
||||
)
|
||||
)
|
||||
return result.scalars().all()
|
||||
|
||||
|
||||
async def get_calendar_entries_for_page(session, post_id: int) -> list[CalendarEntry]:
|
||||
"""Return pending calendar entries scoped to a specific page (via Calendar.post_id)."""
|
||||
ident = current_cart_identity()
|
||||
|
||||
filters = [
|
||||
CalendarEntry.deleted_at.is_(None),
|
||||
CalendarEntry.state == "pending",
|
||||
Calendar.post_id == post_id,
|
||||
Calendar.deleted_at.is_(None),
|
||||
]
|
||||
if ident["user_id"] is not None:
|
||||
filters.append(CalendarEntry.user_id == ident["user_id"])
|
||||
else:
|
||||
filters.append(CalendarEntry.session_id == ident["session_id"])
|
||||
|
||||
result = await session.execute(
|
||||
select(CalendarEntry)
|
||||
.join(Calendar, CalendarEntry.calendar_id == Calendar.id)
|
||||
.where(*filters)
|
||||
.order_by(CalendarEntry.start_at.asc())
|
||||
.options(selectinload(CalendarEntry.calendar))
|
||||
)
|
||||
return result.scalars().all()
|
||||
|
||||
|
||||
async def get_cart_grouped_by_page(session) -> list[dict]:
|
||||
"""
|
||||
Load all cart items + calendar entries for the current identity,
|
||||
grouped by owning page (post_id).
|
||||
|
||||
Returns a list of dicts:
|
||||
{
|
||||
"post": Post | None,
|
||||
"page_config": PageConfig | None,
|
||||
"cart_items": [...],
|
||||
"calendar_entries": [...],
|
||||
"product_count": int,
|
||||
"product_total": float,
|
||||
"calendar_count": int,
|
||||
"calendar_total": float,
|
||||
"total": float,
|
||||
}
|
||||
|
||||
Items without a market_place go in an orphan bucket (post=None).
|
||||
"""
|
||||
from .get_cart import get_cart
|
||||
from .calendar_cart import get_calendar_cart_entries
|
||||
from .total import total as calc_product_total
|
||||
from .calendar_cart import calendar_total as calc_calendar_total
|
||||
|
||||
cart_items = await get_cart(session)
|
||||
cal_entries = await get_calendar_cart_entries(session)
|
||||
|
||||
# Group by post_id
|
||||
groups: dict[int | None, dict] = defaultdict(lambda: {
|
||||
"post_id": None,
|
||||
"cart_items": [],
|
||||
"calendar_entries": [],
|
||||
})
|
||||
|
||||
for ci in cart_items:
|
||||
if ci.market_place and ci.market_place.post_id:
|
||||
pid = ci.market_place.post_id
|
||||
else:
|
||||
pid = None
|
||||
groups[pid]["post_id"] = pid
|
||||
groups[pid]["cart_items"].append(ci)
|
||||
|
||||
for ce in cal_entries:
|
||||
if ce.calendar and ce.calendar.post_id:
|
||||
pid = ce.calendar.post_id
|
||||
else:
|
||||
pid = None
|
||||
groups[pid]["post_id"] = pid
|
||||
groups[pid]["calendar_entries"].append(ce)
|
||||
|
||||
# Batch-load Post and PageConfig objects
|
||||
post_ids = [pid for pid in groups if pid is not None]
|
||||
posts_by_id: dict[int, Post] = {}
|
||||
configs_by_post: dict[int, PageConfig] = {}
|
||||
|
||||
if post_ids:
|
||||
post_result = await session.execute(
|
||||
select(Post).where(Post.id.in_(post_ids))
|
||||
)
|
||||
for p in post_result.scalars().all():
|
||||
posts_by_id[p.id] = p
|
||||
|
||||
pc_result = await session.execute(
|
||||
select(PageConfig).where(PageConfig.post_id.in_(post_ids))
|
||||
)
|
||||
for pc in pc_result.scalars().all():
|
||||
configs_by_post[pc.post_id] = pc
|
||||
|
||||
# Build result list (pages first, orphan last)
|
||||
result = []
|
||||
for pid in sorted(groups, key=lambda x: (x is None, x)):
|
||||
grp = groups[pid]
|
||||
items = grp["cart_items"]
|
||||
entries = grp["calendar_entries"]
|
||||
prod_total = calc_product_total(items) or 0
|
||||
cal_total = calc_calendar_total(entries) or 0
|
||||
|
||||
result.append({
|
||||
"post": posts_by_id.get(pid) if pid else None,
|
||||
"page_config": configs_by_post.get(pid) if pid else None,
|
||||
"cart_items": items,
|
||||
"calendar_entries": entries,
|
||||
"product_count": sum(ci.quantity for ci in items),
|
||||
"product_total": prod_total,
|
||||
"calendar_count": len(entries),
|
||||
"calendar_total": cal_total,
|
||||
"total": prod_total + cal_total,
|
||||
})
|
||||
|
||||
return result
|
||||
@@ -72,12 +72,12 @@ def register() -> Blueprint:
|
||||
return redirect(order.sumup_hosted_url)
|
||||
|
||||
# Otherwise, create a fresh checkout for this order
|
||||
redirect_url = url_for("cart.checkout_return", order_id=order.id, _external=True)
|
||||
redirect_url = url_for("cart_global.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("cart.checkout_webhook", order_id=order.id, _external=True)
|
||||
webhook_url = url_for("cart_global.checkout_webhook", order_id=order.id, _external=True)
|
||||
if webhook_secret:
|
||||
from urllib.parse import urlencode
|
||||
|
||||
|
||||
Reference in New Issue
Block a user