70 lines
2.0 KiB
Python
70 lines
2.0 KiB
Python
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)
|