sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: implement persistent cart with database storage and proper exception handling

Browse Source 
Implemented a complete shopping cart system with database persistence,
replacing the previous stub implementation. The cart now properly stores
items across sessions and follows the project's architecture patterns.

Database Changes:
- Add cart_items table with vendor_id, session_id, product_id, quantity, price_at_add
- Create unique constraint to prevent duplicate items per session
- Add indexes for session lookups and old cart cleanup
- Run migration a2064e1dfcd4 to create cart_items table

New Models & Schemas:
- models/database/cart.py: CartItem SQLAlchemy model with relationships
- models/schema/cart.py: Pydantic schemas for requests/responses
  * AddToCartRequest, UpdateCartItemRequest
  * CartResponse, CartItemResponse, CartOperationResponse, ClearCartResponse

Exception Handling:
- app/exceptions/cart.py: Cart-specific exceptions following project patterns
  * CartItemNotFoundException - item not found in cart
  * InsufficientInventoryForCartException - not enough inventory for cart operation
  * InvalidCartQuantityException - invalid quantity validation
  * CartValidationException - general cart validation
  * EmptyCartException - operations on empty cart
  * ProductNotAvailableForCartException - product unavailable
- Updated app/exceptions/__init__.py to export cart exceptions

Service Layer:
- Implement cart_service.get_cart() - fetch cart from database with totals
- Implement cart_service.add_to_cart() - create or update cart items with inventory checks
- Implement cart_service.update_cart_item() - update quantity with validation
- Implement cart_service.remove_from_cart() - delete cart item
- Implement cart_service.clear_cart() - remove all items for session
- Replace generic exceptions with cart-specific ones
- Fix InsufficientInventoryException usage (was using wrong parameters)

API Layer:
- Update app/api/v1/shop/cart.py to use Pydantic schemas
- Add response_model declarations to all endpoints
- Add return type hints for type safety
- Convert service dict responses to Pydantic models

Features:
- Cart items persist in database across server restarts
- Inventory validation before adding/updating items
- Price captured at time of adding to cart
- Duplicate items update quantity instead of creating new entries
- Full CRUD operations with proper error handling
- Type-safe API with auto-generated OpenAPI documentation

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2025-11-23 20:17:16 +01:00
parent cf2aa2c5af
commit c100d839f1
9 changed files with 688 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
alembic/env.py
View File

@@ -152,6 +152,17 @@ try:
except ImportError as e:
print(f" ✗ Customer models failed: {e}")
# ----------------------------------------------------------------------------
# CART MODELS
# ----------------------------------------------------------------------------
try:
from models.database.cart import CartItem
print(" ✓ Cart models imported (1 model)")
print(" - CartItem")
except ImportError as e:
print(f" ✗ Cart models failed: {e}")
# ----------------------------------------------------------------------------
# ORDER MODELS
# ----------------------------------------------------------------------------

54
alembic/versions/a2064e1dfcd4_add_cart_items_table.py Normal file
View File

@@ -0,0 +1,54 @@
"""add cart_items table
Revision ID: a2064e1dfcd4
Revises: f68d8da5315a
Create Date: 2025-11-23 19:52:40.509538
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'a2064e1dfcd4'
down_revision: Union[str, None] = 'f68d8da5315a'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# Create cart_items table
op.create_table(
'cart_items',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('vendor_id', sa.Integer(), nullable=False),
sa.Column('product_id', sa.Integer(), nullable=False),
sa.Column('session_id', sa.String(length=255), nullable=False),
sa.Column('quantity', sa.Integer(), nullable=False),
sa.Column('price_at_add', sa.Float(), nullable=False),
sa.Column('created_at', sa.DateTime(), nullable=True),
sa.Column('updated_at', sa.DateTime(), nullable=True),
sa.ForeignKeyConstraint(['product_id'], ['products.id'], ),
sa.ForeignKeyConstraint(['vendor_id'], ['vendors.id'], ),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('vendor_id', 'session_id', 'product_id', name='uq_cart_item')
)
# Create indexes
op.create_index('idx_cart_session', 'cart_items', ['vendor_id', 'session_id'], unique=False)
op.create_index('idx_cart_created', 'cart_items', ['created_at'], unique=False)
op.create_index(op.f('ix_cart_items_id'), 'cart_items', ['id'], unique=False)
op.create_index(op.f('ix_cart_items_session_id'), 'cart_items', ['session_id'], unique=False)
def downgrade() -> None:
# Drop indexes
op.drop_index(op.f('ix_cart_items_session_id'), table_name='cart_items')
op.drop_index(op.f('ix_cart_items_id'), table_name='cart_items')
op.drop_index('idx_cart_created', table_name='cart_items')
op.drop_index('idx_cart_session', table_name='cart_items')
# Drop table
op.drop_table('cart_items')

79
app/api/v1/shop/cart.py
View File

@@ -10,40 +10,31 @@ No authentication required - uses session ID for cart tracking.
import logging
from fastapi import APIRouter, Depends, Path, Body, Request, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel, Field
from app.core.database import get_db
from app.services.cart_service import cart_service
from models.schema.cart import (
AddToCartRequest,
UpdateCartItemRequest,
CartResponse,
CartOperationResponse,
ClearCartResponse,
)
router = APIRouter()
logger = logging.getLogger(__name__)
# ============================================================================
# REQUEST/RESPONSE SCHEMAS
# ============================================================================
class AddToCartRequest(BaseModel):
"""Request model for adding to cart."""
product_id: int = Field(..., description="Product ID to add", gt=0)
quantity: int = Field(1, ge=1, description="Quantity to add")
class UpdateCartItemRequest(BaseModel):
"""Request model for updating cart item."""
quantity: int = Field(..., ge=1, description="New quantity")
# ============================================================================
# CART ENDPOINTS
# ============================================================================
@router.get("/cart/{session_id}")
@router.get("/cart/{session_id}", response_model=CartResponse)
def get_cart(
request: Request,
session_id: str = Path(..., description="Shopping session ID"),
db: Session = Depends(get_db),
):
) -> CartResponse:
"""
Get shopping cart contents for current vendor.
@@ -62,8 +53,8 @@ def get_cart(
detail="Vendor not found. Please access via vendor domain/subdomain/path."
)
logger.debug(
f"[SHOP_API] get_cart for session {session_id}",
logger.info(
f"[SHOP_API] get_cart for session {session_id}, vendor {vendor.id}",
extra={
"vendor_id": vendor.id,
"vendor_code": vendor.subdomain,
@@ -77,16 +68,26 @@ def get_cart(
session_id=session_id
)
return cart
logger.info(
f"[SHOP_API] get_cart result: {len(cart.get('items', []))} items in cart",
extra={
"session_id": session_id,
"vendor_id": vendor.id,
"item_count": len(cart.get('items', [])),
"total": cart.get('total', 0),
}
)
return CartResponse.from_service_dict(cart)
@router.post("/cart/{session_id}/items")
@router.post("/cart/{session_id}/items", response_model=CartOperationResponse)
def add_to_cart(
request: Request,
session_id: str = Path(..., description="Shopping session ID"),
cart_data: AddToCartRequest = Body(...),
db: Session = Depends(get_db),
):
) -> CartOperationResponse:
"""
Add product to cart for current vendor.
@@ -109,8 +110,8 @@ def add_to_cart(
detail="Vendor not found. Please access via vendor domain/subdomain/path."
)
logger.debug(
f"[SHOP_API] add_to_cart: product {cart_data.product_id}, qty {cart_data.quantity}",
logger.info(
f"[SHOP_API] add_to_cart: product {cart_data.product_id}, qty {cart_data.quantity}, session {session_id}",
extra={
"vendor_id": vendor.id,
"vendor_code": vendor.subdomain,
@@ -128,17 +129,25 @@ def add_to_cart(
quantity=cart_data.quantity
)
return result
logger.info(
f"[SHOP_API] add_to_cart result: {result}",
extra={
"session_id": session_id,
"result": result,
}
)
return CartOperationResponse(**result)
@router.put("/cart/{session_id}/items/{product_id}")
@router.put("/cart/{session_id}/items/{product_id}", response_model=CartOperationResponse)
def update_cart_item(
request: Request,
session_id: str = Path(..., description="Shopping session ID"),
product_id: int = Path(..., description="Product ID", gt=0),
cart_data: UpdateCartItemRequest = Body(...),
db: Session = Depends(get_db),
):
) -> CartOperationResponse:
"""
Update cart item quantity for current vendor.
@@ -180,16 +189,16 @@ def update_cart_item(
quantity=cart_data.quantity
)
return result
return CartOperationResponse(**result)
@router.delete("/cart/{session_id}/items/{product_id}")
@router.delete("/cart/{session_id}/items/{product_id}", response_model=CartOperationResponse)
def remove_from_cart(
request: Request,
session_id: str = Path(..., description="Shopping session ID"),
product_id: int = Path(..., description="Product ID", gt=0),
db: Session = Depends(get_db),
):
) -> CartOperationResponse:
"""
Remove item from cart for current vendor.
@@ -226,15 +235,15 @@ def remove_from_cart(
product_id=product_id
)
return result
return CartOperationResponse(**result)
@router.delete("/cart/{session_id}")
@router.delete("/cart/{session_id}", response_model=ClearCartResponse)
def clear_cart(
request: Request,
session_id: str = Path(..., description="Shopping session ID"),
db: Session = Depends(get_db),
):
) -> ClearCartResponse:
"""
Clear all items from cart for current vendor.
@@ -268,4 +277,4 @@ def clear_cart(
session_id=session_id
)
return result
return ClearCartResponse(**result)

18
app/exceptions/__init__.py
View File

@@ -170,6 +170,16 @@ from .order import (
OrderCannotBeCancelledException,
)
# Cart exceptions
from .cart import (
CartItemNotFoundException,
EmptyCartException,
CartValidationException,
InsufficientInventoryForCartException,
InvalidCartQuantityException,
ProductNotAvailableForCartException,
)
__all__ = [
# Base exceptions
"WizamartException",
@@ -277,6 +287,14 @@ __all__ = [
"InvalidOrderStatusException",
"OrderCannotBeCancelledException",
# Cart exceptions
"CartItemNotFoundException",
"EmptyCartException",
"CartValidationException",
"InsufficientInventoryForCartException",
"InvalidCartQuantityException",
"ProductNotAvailableForCartException",
# MarketplaceProduct exceptions
"MarketplaceProductNotFoundException",
"MarketplaceProductAlreadyExistsException",

116
app/exceptions/cart.py Normal file
View File

@@ -0,0 +1,116 @@
# app/exceptions/cart.py
"""
Shopping cart specific exceptions.
"""
from typing import Optional
from .base import (
ResourceNotFoundException,
ValidationException,
BusinessLogicException
)
class CartItemNotFoundException(ResourceNotFoundException):
"""Raised when a cart item is not found."""
def __init__(self, product_id: int, session_id: str):
super().__init__(
resource_type="CartItem",
identifier=str(product_id),
message=f"Product {product_id} not found in cart",
error_code="CART_ITEM_NOT_FOUND"
)
self.details.update({
"product_id": product_id,
"session_id": session_id
})
class EmptyCartException(ValidationException):
"""Raised when trying to perform operations on an empty cart."""
def __init__(self, session_id: str):
super().__init__(
message="Cart is empty",
details={"session_id": session_id}
)
self.error_code = "CART_EMPTY"
class CartValidationException(ValidationException):
"""Raised when cart data validation fails."""
def __init__(
self,
message: str = "Cart validation failed",
field: Optional[str] = None,
details: Optional[dict] = None,
):
super().__init__(
message=message,
field=field,
details=details,
)
self.error_code = "CART_VALIDATION_FAILED"
class InsufficientInventoryForCartException(BusinessLogicException):
"""Raised when product doesn't have enough inventory for cart operation."""
def __init__(
self,
product_id: int,
product_name: str,
requested: int,
available: int,
):
message = f"Insufficient inventory for product '{product_name}'. Requested: {requested}, Available: {available}"
super().__init__(
message=message,
error_code="INSUFFICIENT_INVENTORY_FOR_CART",
details={
"product_id": product_id,
"product_name": product_name,
"requested_quantity": requested,
"available_quantity": available,
},
)
class InvalidCartQuantityException(ValidationException):
"""Raised when cart quantity is invalid."""
def __init__(self, quantity: int, min_quantity: int = 1, max_quantity: Optional[int] = None):
if quantity < min_quantity:
message = f"Quantity must be at least {min_quantity}"
elif max_quantity and quantity > max_quantity:
message = f"Quantity cannot exceed {max_quantity}"
else:
message = f"Invalid quantity: {quantity}"
super().__init__(
message=message,
field="quantity",
details={
"quantity": quantity,
"min_quantity": min_quantity,
"max_quantity": max_quantity,
},
)
self.error_code = "INVALID_CART_QUANTITY"
class ProductNotAvailableForCartException(BusinessLogicException):
"""Raised when product is not available for adding to cart."""
def __init__(self, product_id: int, reason: str):
super().__init__(
message=f"Product {product_id} cannot be added to cart: {reason}",
error_code="PRODUCT_NOT_AVAILABLE_FOR_CART",
details={
"product_id": product_id,
"reason": reason,
},
)

381
app/services/cart_service.py
View File

@@ -17,10 +17,14 @@ from sqlalchemy import and_
from models.database.product import Product
from models.database.vendor import Vendor
from models.database.cart import CartItem
from app.exceptions import (
ProductNotFoundException,
ValidationException,
InsufficientInventoryException
CartItemNotFoundException,
CartValidationException,
InsufficientInventoryForCartException,
InvalidCartQuantityException,
ProductNotAvailableForCartException,
)
logger = logging.getLogger(__name__)
@@ -38,19 +42,69 @@ class CartService:
"""
Get cart contents for a session.
Note: This is a simple in-memory implementation.
In production, you'd store carts in Redis or database.
Args:
db: Database session
vendor_id: Vendor ID
session_id: Session ID
Returns:
Cart data with items and totals
"""
# For now, return empty cart structure
# TODO: Implement persistent cart storage
return {
logger.info(
f"[CART_SERVICE] get_cart called",
extra={
"vendor_id": vendor_id,
"session_id": session_id,
}
)
# Fetch cart items from database
cart_items = db.query(CartItem).filter(
and_(
CartItem.vendor_id == vendor_id,
CartItem.session_id == session_id
)
).all()
logger.info(
f"[CART_SERVICE] Found {len(cart_items)} items in database",
extra={"item_count": len(cart_items)}
)
# Build response
items = []
subtotal = 0.0
for cart_item in cart_items:
product = cart_item.product
line_total = cart_item.line_total
items.append({
"product_id": product.id,
"product_name": product.name,
"quantity": cart_item.quantity,
"price": cart_item.price_at_add,
"line_total": line_total,
"image_url": product.image_link if hasattr(product, 'image_link') else None,
})
subtotal += line_total
cart_data = {
"vendor_id": vendor_id,
"session_id": session_id,
"items": [],
"subtotal": 0.0,
"total": 0.0
"items": items,
"subtotal": subtotal,
"total": subtotal # Could add tax/shipping later
}
logger.info(
f"[CART_SERVICE] get_cart returning: {len(cart_data['items'])} items, total: {cart_data['total']}",
extra={"cart": cart_data}
)
return cart_data
def add_to_cart(
self,
db: Session,
@@ -76,6 +130,16 @@ class CartService:
ProductNotFoundException: If product not found
InsufficientInventoryException: If not enough inventory
"""
logger.info(
f"[CART_SERVICE] add_to_cart called",
extra={
"vendor_id": vendor_id,
"session_id": session_id,
"product_id": product_id,
"quantity": quantity
}
)
# Verify product exists and belongs to vendor
product = db.query(Product).filter(
and_(
@@ -85,64 +149,195 @@ class CartService:
)
).first()
if not product:
logger.error(
f"[CART_SERVICE] Product not found",
extra={"product_id": product_id, "vendor_id": vendor_id}
)
raise ProductNotFoundException(product_id=product_id, vendor_id=vendor_id)
logger.info(
f"[CART_SERVICE] Product found: {product.name}",
extra={
"product_id": product_id,
"product_name": product.name,
"available_inventory": product.available_inventory
}
)
# Get current price (use sale_price if available, otherwise regular price)
current_price = product.sale_price if product.sale_price else product.price
# Check if item already exists in cart
existing_item = db.query(CartItem).filter(
and_(
CartItem.vendor_id == vendor_id,
CartItem.session_id == session_id,
CartItem.product_id == product_id
)
).first()
if existing_item:
# Update quantity
new_quantity = existing_item.quantity + quantity
# Check inventory for new total quantity
if product.available_inventory < new_quantity:
logger.warning(
f"[CART_SERVICE] Insufficient inventory for update",
extra={
"product_id": product_id,
"current_in_cart": existing_item.quantity,
"adding": quantity,
"requested_total": new_quantity,
"available": product.available_inventory
}
)
raise InsufficientInventoryForCartException(
product_id=product_id,
product_name=product.name,
requested=new_quantity,
available=product.available_inventory
)
existing_item.quantity = new_quantity
db.commit()
db.refresh(existing_item)
logger.info(
f"[CART_SERVICE] Updated existing cart item",
extra={
"cart_item_id": existing_item.id,
"new_quantity": new_quantity
}
)
return {
"message": "Product quantity updated in cart",
"product_id": product_id,
"quantity": new_quantity
}
else:
# Check inventory for new item
if product.available_inventory < quantity:
logger.warning(
f"[CART_SERVICE] Insufficient inventory",
extra={
"product_id": product_id,
"requested": quantity,
"available": product.available_inventory
}
)
raise InsufficientInventoryForCartException(
product_id=product_id,
product_name=product.name,
requested=quantity,
available=product.available_inventory
)
# Create new cart item
cart_item = CartItem(
vendor_id=vendor_id,
session_id=session_id,
product_id=product_id,
quantity=quantity,
price_at_add=current_price
)
db.add(cart_item)
db.commit()
db.refresh(cart_item)
logger.info(
f"[CART_SERVICE] Created new cart item",
extra={
"cart_item_id": cart_item.id,
"quantity": quantity,
"price": current_price
}
)
return {
"message": "Product added to cart",
"product_id": product_id,
"quantity": quantity
}
def update_cart_item(
self,
db: Session,
vendor_id: int,
session_id: str,
product_id: int,
quantity: int
) -> Dict:
"""
Update quantity of item in cart.
Args:
db: Database session
vendor_id: Vendor ID
session_id: Session ID
product_id: Product ID
quantity: New quantity (must be >= 1)
Returns:
Success message
Raises:
ValidationException: If quantity < 1
ProductNotFoundException: If product not found
InsufficientInventoryException: If not enough inventory
"""
if quantity < 1:
raise InvalidCartQuantityException(quantity=quantity, min_quantity=1)
# Find cart item
cart_item = db.query(CartItem).filter(
and_(
CartItem.vendor_id == vendor_id,
CartItem.session_id == session_id,
CartItem.product_id == product_id
)
).first()
if not cart_item:
raise CartItemNotFoundException(product_id=product_id, session_id=session_id)
# Verify product still exists and is active
product = db.query(Product).filter(
and_(
Product.id == product_id,
Product.vendor_id == vendor_id,
Product.is_active == True
)
).first()
if not product:
raise ProductNotFoundException(str(product_id))
# Check inventory
if product.available_inventory < quantity:
raise InsufficientInventoryException(
raise InsufficientInventoryForCartException(
product_id=product_id,
product_name=product.name,
requested=quantity,
available=product.available_inventory
)
# TODO: Add to persistent cart storage
# For now, return success response
# Update quantity
cart_item.quantity = quantity
db.commit()
db.refresh(cart_item)
logger.info(
f"Added product {product_id} (qty: {quantity}) to cart "
f"for session {session_id}"
f"[CART_SERVICE] Updated cart item quantity",
extra={
"cart_item_id": cart_item.id,
"product_id": product_id,
"new_quantity": quantity
}
)
return {
"message": "Product added to cart",
"product_id": product_id,
"quantity": quantity
}
def update_cart_item(
self,
db: Session,
vendor_id: int,
session_id: str,
product_id: int,
quantity: int
) -> Dict:
"""Update quantity of item in cart."""
if quantity < 1:
raise ValidationException("Quantity must be at least 1")
# Verify product
product = db.query(Product).filter(
and_(
Product.id == product_id,
Product.vendor_id == vendor_id
)
).first()
if not product:
raise ProductNotFoundException(str(product_id))
# Check inventory
if product.available_inventory < quantity:
raise InsufficientInventoryException(
product_id=product_id,
requested=quantity,
available=product.available_inventory
)
# TODO: Update persistent cart
logger.info(f"Updated cart item {product_id} quantity to {quantity}")
return {
"message": "Cart updated",
"product_id": product_id,
@@ -156,9 +351,44 @@ class CartService:
session_id: str,
product_id: int
) -> Dict:
"""Remove item from cart."""
# TODO: Remove from persistent cart
logger.info(f"Removed product {product_id} from cart {session_id}")
"""
Remove item from cart.
Args:
db: Database session
vendor_id: Vendor ID
session_id: Session ID
product_id: Product ID
Returns:
Success message
Raises:
ProductNotFoundException: If product not in cart
"""
# Find and delete cart item
cart_item = db.query(CartItem).filter(
and_(
CartItem.vendor_id == vendor_id,
CartItem.session_id == session_id,
CartItem.product_id == product_id
)
).first()
if not cart_item:
raise CartItemNotFoundException(product_id=product_id, session_id=session_id)
db.delete(cart_item)
db.commit()
logger.info(
f"[CART_SERVICE] Removed item from cart",
extra={
"cart_item_id": cart_item.id,
"product_id": product_id,
"session_id": session_id
}
)
return {
"message": "Item removed from cart",
@@ -171,12 +401,39 @@ class CartService:
vendor_id: int,
session_id: str
) -> Dict:
"""Clear all items from cart."""
# TODO: Clear persistent cart
logger.info(f"Cleared cart for session {session_id}")
"""
Clear all items from cart.
Args:
db: Database session
vendor_id: Vendor ID
session_id: Session ID
Returns:
Success message with count of items removed
"""
# Delete all cart items for this session
deleted_count = db.query(CartItem).filter(
and_(
CartItem.vendor_id == vendor_id,
CartItem.session_id == session_id
)
).delete()
db.commit()
logger.info(
f"[CART_SERVICE] Cleared cart",
extra={
"session_id": session_id,
"vendor_id": vendor_id,
"items_removed": deleted_count
}
)
return {
"message": "Cart cleared"
"message": "Cart cleared",
"items_removed": deleted_count
}

46
models/database/cart.py Normal file
View File

@@ -0,0 +1,46 @@
# models/database/cart.py
"""Cart item database model."""
from datetime import datetime
from sqlalchemy import Column, Float, ForeignKey, Index, Integer, String, UniqueConstraint
from sqlalchemy.orm import relationship
from app.core.database import Base
from models.database.base import TimestampMixin
class CartItem(Base, TimestampMixin):
"""
Shopping cart items.
Stores cart items per session, vendor, and product.
Sessions are identified by a session_id string (from browser cookies).
"""
__tablename__ = "cart_items"
id = Column(Integer, primary_key=True, index=True)
vendor_id = Column(Integer, ForeignKey("vendors.id"), nullable=False)
product_id = Column(Integer, ForeignKey("products.id"), nullable=False)
session_id = Column(String(255), nullable=False, index=True)
# Cart details
quantity = Column(Integer, nullable=False, default=1)
price_at_add = Column(Float, nullable=False) # Store price when added to cart
# Relationships
vendor = relationship("Vendor")
product = relationship("Product")
# Constraints
__table_args__ = (
UniqueConstraint("vendor_id", "session_id", "product_id", name="uq_cart_item"),
Index("idx_cart_session", "vendor_id", "session_id"),
Index("idx_cart_created", "created_at"), # For cleanup of old carts
)
def __repr__(self):
return f"<CartItem(id={self.id}, session='{self.session_id}', product_id={self.product_id}, qty={self.quantity})>"
@property
def line_total(self) -> float:
"""Calculate line total."""
return self.price_at_add * self.quantity

80
models/schema/cart.py Normal file
View File

@@ -0,0 +1,80 @@
# models/schema/cart.py
"""
Pydantic schemas for shopping cart operations.
"""
from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel, Field, ConfigDict
# ============================================================================
# Request Schemas
# ============================================================================
class AddToCartRequest(BaseModel):
"""Request model for adding items to cart."""
product_id: int = Field(..., description="Product ID to add", gt=0)
quantity: int = Field(1, ge=1, description="Quantity to add")
class UpdateCartItemRequest(BaseModel):
"""Request model for updating cart item quantity."""
quantity: int = Field(..., ge=1, description="New quantity (must be >= 1)")
# ============================================================================
# Response Schemas
# ============================================================================
class CartItemResponse(BaseModel):
"""Response model for a single cart item."""
model_config = ConfigDict(from_attributes=True)
product_id: int = Field(..., description="Product ID")
product_name: str = Field(..., description="Product name")
quantity: int = Field(..., description="Quantity in cart")
price: float = Field(..., description="Price per unit when added to cart")
line_total: float = Field(..., description="Total price for this line (price * quantity)")
image_url: Optional[str] = Field(None, description="Product image URL")
class CartResponse(BaseModel):
"""Response model for shopping cart."""
vendor_id: int = Field(..., description="Vendor ID")
session_id: str = Field(..., description="Shopping session ID")
items: List[CartItemResponse] = Field(default_factory=list, description="Cart items")
subtotal: float = Field(..., description="Subtotal of all items")
total: float = Field(..., description="Total amount (currently same as subtotal)")
item_count: int = Field(..., description="Total number of items in cart")
@classmethod
def from_service_dict(cls, cart_dict: dict) -> "CartResponse":
"""
Create CartResponse from service layer dictionary.
This is a convenience method to convert the dictionary format
returned by cart_service into a proper Pydantic model.
"""
items = [CartItemResponse(**item) for item in cart_dict.get("items", [])]
return cls(
vendor_id=cart_dict["vendor_id"],
session_id=cart_dict["session_id"],
items=items,
subtotal=cart_dict["subtotal"],
total=cart_dict["total"],
item_count=len(items)
)
class CartOperationResponse(BaseModel):
"""Response model for cart operations (add, update, remove)."""
message: str = Field(..., description="Operation result message")
product_id: int = Field(..., description="Product ID affected")
quantity: Optional[int] = Field(None, description="New quantity (for add/update operations)")
class ClearCartResponse(BaseModel):
"""Response model for clearing cart."""
message: str = Field(..., description="Operation result message")
items_removed: int = Field(..., description="Number of items removed from cart")

0
static/shared/static/shared/js/vendor/alpine.min.js → static/shared/js/vendor/alpine.min.js vendored
View File