giles
fa431ee13e
Some checks failed
Build and Deploy / build-and-deploy (push) Has been cancelled
Phase 1 - Relations service (internal): owns ContainerRelation, exposes get-children data + attach/detach-child actions. Retargeted events, blog, market callers from cart to relations. Phase 2 - Likes service (internal): unified Like model replaces ProductLike and PostLike with generic target_type/target_slug/target_id. Exposes is-liked, liked-slugs, liked-ids data + toggle action. Phase 3 - PageConfig → blog: moved ownership to blog with direct DB queries, removed proxy endpoints from cart. Phase 4 - Orders service (public): owns Order/OrderItem + SumUp checkout flow. Cart checkout now delegates to orders via create-order action. Webhook/return routes and reconciliation moved to orders. Phase 5 - Infrastructure: docker-compose, deploy.sh, Dockerfiles updated for all 3 new services. Added orders_url helper and factory model imports. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
139 lines
4.5 KiB
Python
139 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
from quart import Blueprint, g, render_template, redirect, url_for, make_response
|
|
from sqlalchemy import select, func, or_, cast, String, exists
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from shared.models.order import Order, OrderItem
|
|
|
|
from shared.infrastructure.http_utils import vary as _vary, current_url_without_page as _current_url_without_page
|
|
from shared.infrastructure.cart_identity import current_cart_identity
|
|
from shared.browser.app.utils.htmx import is_htmx_request
|
|
from bp.order.routes import register as register_order
|
|
|
|
from .filters.qs import makeqs_factory, decode
|
|
|
|
|
|
def register(url_prefix: str) -> Blueprint:
|
|
bp = Blueprint("orders", __name__, url_prefix=url_prefix)
|
|
bp.register_blueprint(register_order())
|
|
|
|
ORDERS_PER_PAGE = 10
|
|
|
|
oob = {
|
|
"extends": "_types/root/_index.html",
|
|
"child_id": "auth-header-child",
|
|
"header": "_types/auth/header/_header.html",
|
|
"nav": "_types/auth/_nav.html",
|
|
"main": "_types/auth/_main_panel.html",
|
|
}
|
|
|
|
@bp.context_processor
|
|
def inject_oob():
|
|
return {"oob": oob}
|
|
|
|
@bp.before_request
|
|
def route():
|
|
g.makeqs_factory = makeqs_factory
|
|
|
|
@bp.before_request
|
|
async def _require_identity():
|
|
"""Orders require a logged-in user or at least a cart session."""
|
|
ident = current_cart_identity()
|
|
if not ident["user_id"] and not ident["session_id"]:
|
|
return redirect(url_for("auth.login_form"))
|
|
|
|
@bp.get("/")
|
|
async def list_orders():
|
|
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:
|
|
return redirect(url_for("auth.login_form"))
|
|
|
|
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()
|
|
|
|
context = {
|
|
"orders": orders,
|
|
"page": page,
|
|
"total_pages": total_pages,
|
|
"search": search,
|
|
"search_count": total_count,
|
|
}
|
|
|
|
if not is_htmx_request():
|
|
html = await render_template("_types/orders/index.html", **context)
|
|
elif page > 1:
|
|
html = await render_template("_types/orders/_rows.html", **context)
|
|
else:
|
|
html = await render_template("_types/orders/_oob_elements.html", **context)
|
|
|
|
resp = await make_response(html)
|
|
resp.headers["Hx-Push-Url"] = _current_url_without_page()
|
|
return _vary(resp)
|
|
|
|
return bp
|