from typing import List, Optional from sqlalchemy.orm import Session, joinedload from app.crud.base import CRUDBase from app.models.order import Order, OrderItem from app.schemas.order import OrderCreate, OrderItemCreate, OrderUpdate class CRUDOrder(CRUDBase[Order, OrderCreate, OrderUpdate]): def get_orders_by_user( self, db: Session, *, user_id: int, skip: int = 0, limit: int = 100 ) -> List[Order]: return ( db.query(Order) .filter(Order.user_id == user_id) .order_by(Order.created_at.desc()) .offset(skip) .limit(limit) .all() ) def get_order_with_items(self, db: Session, *, order_id: int) -> Optional[Order]: return ( db.query(Order) .filter(Order.id == order_id) .options(joinedload(Order.items)) .first() ) def create_with_items( self, db: Session, *, obj_in: OrderCreate, user_id: int ) -> Order: # Calculate the total amount total_amount = sum(item.unit_price * item.quantity for item in obj_in.items) # Create order db_obj = Order( user_id=user_id, status=obj_in.status, total_amount=total_amount, shipping_address=obj_in.shipping_address, payment_details=obj_in.payment_details, tracking_number=obj_in.tracking_number, ) db.add(db_obj) db.flush() # Create order items for item in obj_in.items: order_item = OrderItem( order_id=db_obj.id, product_id=item.product_id, quantity=item.quantity, unit_price=item.unit_price, ) db.add(order_item) db.commit() db.refresh(db_obj) return db_obj class CRUDOrderItem(CRUDBase[OrderItem, OrderItemCreate, OrderItemCreate]): pass order = CRUDOrder(Order) order_item = CRUDOrderItem(OrderItem)