8527ddb84b
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 50s
- Replace all imports from blog.models, market.models, events.models and bare models.* with shared.models equivalents - Convert cart/models/order.py and page_config.py to re-export stubs - Update shared + glue submodule pointers Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
38 lines
1.3 KiB
Python
38 lines
1.3 KiB
Python
from sqlalchemy import update, func, select
|
||
|
||
from shared.models.market import CartItem
|
||
from shared.models.market_place import MarketPlace
|
||
from shared.models.order import Order
|
||
|
||
|
||
async def clear_cart_for_order(session, order: Order, *, page_post_id: int | None = None) -> None:
|
||
"""
|
||
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:
|
||
filters.append(CartItem.user_id == order.user_id)
|
||
if order.session_id is not None:
|
||
filters.append(CartItem.session_id == order.session_id)
|
||
|
||
if len(filters) == 1:
|
||
# 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.container_type == "page",
|
||
MarketPlace.container_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)
|
||
.values(deleted_at=func.now())
|
||
)
|