146 lines
4.1 KiB
Python
146 lines
4.1 KiB
Python
from typing import List, Dict, Optional, Union, Any, Tuple
|
|
from decimal import Decimal
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.exc import IntegrityError
|
|
from models.order import Order
|
|
from models.user import User
|
|
from datetime import datetime
|
|
import re
|
|
|
|
def validate_order_number(order_number: str) -> bool:
|
|
"""
|
|
Validate order number format (alphanumeric, length between 8-12 chars).
|
|
|
|
Args:
|
|
order_number: The order number to validate
|
|
|
|
Returns:
|
|
bool: True if valid format, False otherwise
|
|
"""
|
|
pattern = r'^[A-Z0-9]{8,12}$'
|
|
return bool(re.match(pattern, order_number))
|
|
|
|
def calculate_order_total(items: List[Dict[str, Any]]) -> Decimal:
|
|
"""
|
|
Calculate total order amount from list of items.
|
|
|
|
Args:
|
|
items: List of order items with quantity and price
|
|
|
|
Returns:
|
|
Decimal: Total order amount
|
|
"""
|
|
total = Decimal('0.00')
|
|
for item in items:
|
|
quantity = Decimal(str(item.get('quantity', 0)))
|
|
price = Decimal(str(item.get('price', 0)))
|
|
total += quantity * price
|
|
return total.quantize(Decimal('0.01'))
|
|
|
|
def create_order_safely(
|
|
db: Session,
|
|
customer_id: str,
|
|
items: List[Dict[str, Any]],
|
|
shipping_address: str
|
|
) -> Tuple[Optional[Order], Optional[str]]:
|
|
"""
|
|
Create new order with validation and error handling.
|
|
|
|
Args:
|
|
db: Database session
|
|
customer_id: ID of customer placing order
|
|
items: List of order items
|
|
shipping_address: Shipping address
|
|
|
|
Returns:
|
|
Tuple containing Order object if successful and None if error,
|
|
or None and error message if failed
|
|
"""
|
|
try:
|
|
# Validate customer exists
|
|
customer = db.query(User).filter(User.id == customer_id).first()
|
|
if not customer:
|
|
return None, "Customer not found"
|
|
|
|
# Generate unique order number
|
|
order_number = generate_unique_order_number(db)
|
|
|
|
# Calculate total
|
|
total_amount = calculate_order_total(items)
|
|
|
|
order = Order(
|
|
order_number=order_number,
|
|
customer_id=customer_id,
|
|
total_amount=total_amount,
|
|
shipping_address=shipping_address,
|
|
status="pending",
|
|
payment_status="unpaid"
|
|
)
|
|
|
|
db.add(order)
|
|
db.commit()
|
|
db.refresh(order)
|
|
|
|
return order, None
|
|
|
|
except IntegrityError:
|
|
db.rollback()
|
|
return None, "Error creating order"
|
|
except Exception as e:
|
|
db.rollback()
|
|
return None, str(e)
|
|
|
|
def generate_unique_order_number(db: Session) -> str:
|
|
"""
|
|
Generate unique order number.
|
|
|
|
Args:
|
|
db: Database session
|
|
|
|
Returns:
|
|
str: Unique order number
|
|
"""
|
|
while True:
|
|
timestamp = datetime.now().strftime('%Y%m%d')
|
|
random_chars = ''.join(random.choices('ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789', k=4))
|
|
order_number = f"ORD{timestamp}{random_chars}"
|
|
|
|
exists = db.query(Order).filter(Order.order_number == order_number).first()
|
|
if not exists:
|
|
return order_number
|
|
|
|
def update_order_status(
|
|
db: Session,
|
|
order_number: str,
|
|
new_status: str,
|
|
new_payment_status: Optional[str] = None
|
|
) -> Tuple[Optional[Order], Optional[str]]:
|
|
"""
|
|
Update order status and optionally payment status.
|
|
|
|
Args:
|
|
db: Database session
|
|
order_number: Order number to update
|
|
new_status: New order status
|
|
new_payment_status: Optional new payment status
|
|
|
|
Returns:
|
|
Tuple containing updated Order if successful and None if error,
|
|
or None and error message if failed
|
|
"""
|
|
try:
|
|
order = db.query(Order).filter(Order.order_number == order_number).first()
|
|
if not order:
|
|
return None, "Order not found"
|
|
|
|
order.status = new_status
|
|
if new_payment_status:
|
|
order.payment_status = new_payment_status
|
|
|
|
db.commit()
|
|
db.refresh(order)
|
|
return order, None
|
|
|
|
except Exception as e:
|
|
db.rollback()
|
|
return None, str(e) |