609e7fb237
This API provides endpoints for: - Product and inventory management - Customer management - Order processing - Payment processing with Stripe integration - User authentication generated with BackendIM... (backend.im)
77 lines
2.9 KiB
Python
77 lines
2.9 KiB
Python
import uuid
|
|
from typing import List, Optional
|
|
|
|
from sqlalchemy import desc
|
|
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, OrderUpdate, OrderItemCreate
|
|
|
|
|
|
class CRUDOrder(CRUDBase[Order, OrderCreate, OrderUpdate]):
|
|
def get_by_order_number(self, db: Session, *, order_number: str) -> Optional[Order]:
|
|
return db.query(Order).filter(Order.order_number == order_number).first()
|
|
|
|
def get_by_customer(self, db: Session, *, customer_id: int, skip: int = 0, limit: int = 100) -> List[Order]:
|
|
return db.query(Order).filter(Order.customer_id == customer_id).order_by(desc(Order.created_at)).offset(skip).limit(limit).all()
|
|
|
|
def get_by_payment_intent(self, db: Session, *, payment_intent_id: str) -> Optional[Order]:
|
|
return db.query(Order).filter(Order.payment_intent_id == payment_intent_id).first()
|
|
|
|
def get_with_items(self, db: Session, *, order_id: int) -> Optional[Order]:
|
|
return db.query(Order).options(joinedload(Order.items)).filter(Order.id == order_id).first()
|
|
|
|
def create_with_items(self, db: Session, *, obj_in: OrderCreate) -> Order:
|
|
# Generate unique order number
|
|
order_number = f"ORD-{uuid.uuid4().hex[:8].upper()}"
|
|
|
|
# Create the order
|
|
db_obj = Order(
|
|
customer_id=obj_in.customer_id,
|
|
order_number=order_number,
|
|
status=obj_in.status,
|
|
total_amount=obj_in.total_amount,
|
|
payment_status="unpaid",
|
|
shipping_address=obj_in.shipping_address,
|
|
shipping_city=obj_in.shipping_city,
|
|
shipping_state=obj_in.shipping_state,
|
|
shipping_zip=obj_in.shipping_zip,
|
|
shipping_country=obj_in.shipping_country
|
|
)
|
|
db.add(db_obj)
|
|
db.flush() # Flush to get the order id
|
|
|
|
# Create the order items
|
|
for item_in in obj_in.items:
|
|
order_item = OrderItem(
|
|
order_id=db_obj.id,
|
|
product_id=item_in.product_id,
|
|
quantity=item_in.quantity,
|
|
unit_price=item_in.unit_price
|
|
)
|
|
db.add(order_item)
|
|
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def update_payment_status(self, db: Session, *, order_id: int, payment_intent_id: str, payment_status: str) -> Order:
|
|
order = self.get(db, id=order_id)
|
|
if order:
|
|
order.payment_intent_id = payment_intent_id
|
|
order.payment_status = payment_status
|
|
|
|
# Update order status based on payment status
|
|
if payment_status == "paid":
|
|
order.status = "processing"
|
|
elif payment_status == "refunded":
|
|
order.status = "cancelled"
|
|
|
|
db.add(order)
|
|
db.commit()
|
|
db.refresh(order)
|
|
return order
|
|
|
|
|
|
order = CRUDOrder(Order) |