from typing import List, Optional from datetime import datetime, timedelta from uuid import uuid4 from sqlalchemy.orm import Session, joinedload from app.models.cart import Cart, CartItem from app.schemas.cart import CartCreate, CartUpdate, CartItemCreate, CartItemUpdate from app.core.config import settings from app.crud.base import CRUDBase class CRUDCartItem(CRUDBase[CartItem, CartItemCreate, CartItemUpdate]): """CRUD operations for CartItem model.""" def create_with_cart( self, db: Session, *, obj_in: CartItemCreate, cart_id: int, product_price: float ) -> CartItem: """Create a new cart item with cart ID and product price.""" cart_item = CartItem( cart_id=cart_id, product_id=obj_in.product_id, quantity=obj_in.quantity, unit_price=product_price ) db.add(cart_item) db.commit() db.refresh(cart_item) return cart_item def get_by_cart_and_product( self, db: Session, *, cart_id: int, product_id: int ) -> Optional[CartItem]: """Get a cart item by cart ID and product ID.""" return db.query(CartItem).filter( CartItem.cart_id == cart_id, CartItem.product_id == product_id ).first() def get_by_cart(self, db: Session, *, cart_id: int) -> List[CartItem]: """Get all items in a cart.""" return db.query(CartItem).filter(CartItem.cart_id == cart_id).all() class CRUDCart(CRUDBase[Cart, CartCreate, CartUpdate]): """CRUD operations for Cart model.""" def create_with_owner( self, db: Session, *, obj_in: CartCreate, user_id: Optional[int] = None ) -> Cart: """Create a new cart with owner.""" session_id = None if user_id else str(uuid4()) expires_at = datetime.utcnow() + timedelta(hours=settings.CART_EXPIRATION_HOURS) cart = Cart( user_id=user_id, session_id=session_id, is_active=True, expires_at=expires_at ) db.add(cart) db.commit() db.refresh(cart) return cart def get_active_by_user(self, db: Session, *, user_id: int) -> Optional[Cart]: """Get the active cart for a user.""" return db.query(Cart).filter( Cart.user_id == user_id, Cart.is_active ).first() def get_active_by_session(self, db: Session, *, session_id: str) -> Optional[Cart]: """Get the active cart for a session.""" return db.query(Cart).filter( Cart.session_id == session_id, Cart.is_active ).first() def get_or_create_active_cart( self, db: Session, *, user_id: Optional[int] = None, session_id: Optional[str] = None ) -> Cart: """Get or create an active cart for a user or session.""" # Try to get existing active cart cart = None if user_id: cart = self.get_active_by_user(db, user_id=user_id) elif session_id: cart = self.get_active_by_session(db, session_id=session_id) # Create new cart if none exists if not cart: obj_in = CartCreate(user_id=user_id, session_id=session_id) cart = self.create_with_owner(db, obj_in=obj_in, user_id=user_id) return cart def get_cart_with_items(self, db: Session, *, cart_id: int) -> Optional[Cart]: """Get a cart with all its items and product details.""" return db.query(Cart).options( joinedload(Cart.items).joinedload(CartItem.product) ).filter(Cart.id == cart_id).first() def clean_expired_carts(self, db: Session) -> int: """Clean up expired carts. Returns count of removed carts.""" now = datetime.utcnow() expired_carts = db.query(Cart).filter( Cart.expires_at < now, Cart.is_active ).all() count = len(expired_carts) for cart in expired_carts: cart.is_active = False db.add(cart) db.commit() return count cart = CRUDCart(Cart) cart_item = CRUDCartItem(CartItem)