d2b80dacc4
- Set up FastAPI application structure - Create database models for User, Product, Cart, CartItem, Order, and OrderItem - Set up Alembic for database migrations - Create Pydantic schemas for request/response models - Implement API endpoints for products, cart operations, and checkout process - Add health endpoint - Update README with project details and documentation
122 lines
4.1 KiB
Python
122 lines
4.1 KiB
Python
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) |