47 lines
1.5 KiB
Python
47 lines
1.5 KiB
Python
from typing import List
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.crud.base import CRUDBase
|
|
from app.models.order import Order, OrderItem
|
|
from app.schemas.order import OrderCreate, OrderUpdate
|
|
|
|
|
|
class CRUDOrder(CRUDBase[Order, OrderCreate, OrderUpdate]):
|
|
def create(self, db: Session, *, obj_in: OrderCreate, user_id: int) -> Order:
|
|
# Calculate total amount from order items
|
|
total_amount = sum(item.unit_price * item.quantity for item in obj_in.order_items)
|
|
|
|
# Create order
|
|
order_data = obj_in.dict(exclude={"order_items"})
|
|
db_obj = Order(**order_data, user_id=user_id, total_amount=total_amount)
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
|
|
# Create order items
|
|
for item in obj_in.order_items:
|
|
db_item = OrderItem(
|
|
order_id=db_obj.id,
|
|
product_id=item.product_id,
|
|
quantity=item.quantity,
|
|
unit_price=item.unit_price
|
|
)
|
|
db.add(db_item)
|
|
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def get_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).offset(skip).limit(limit).all()
|
|
|
|
def get_by_status(
|
|
self, db: Session, *, status: str, skip: int = 0, limit: int = 100
|
|
) -> List[Order]:
|
|
return db.query(Order).filter(Order.status == status).offset(skip).limit(limit).all()
|
|
|
|
|
|
order = CRUDOrder(Order) |