b078a91dd3
- Setup project structure and FastAPI application - Create SQLAlchemy models for users, products, carts, and orders - Implement Alembic migrations - Add CRUD operations and endpoints for all resources - Setup authentication with JWT - Add role-based access control - Update documentation
114 lines
2.9 KiB
Python
114 lines
2.9 KiB
Python
from typing import List, Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.crud.base import CRUDBase
|
|
from app.models.order import Order, OrderItem, OrderStatus
|
|
from app.schemas.order import OrderCreate, OrderUpdate, OrderItemCreate
|
|
|
|
|
|
class CRUDOrder(CRUDBase[Order, OrderCreate, OrderUpdate]):
|
|
def get_user_orders(self, db: Session, *, user_id: int) -> List[Order]:
|
|
"""
|
|
Get all orders for a user.
|
|
|
|
Args:
|
|
db: Database session
|
|
user_id: User ID
|
|
|
|
Returns:
|
|
List of orders
|
|
"""
|
|
return db.query(self.model).filter(self.model.user_id == user_id).all()
|
|
|
|
def create_with_items(
|
|
self,
|
|
db: Session,
|
|
*,
|
|
obj_in: OrderCreate,
|
|
user_id: int,
|
|
items: List[OrderItemCreate],
|
|
total_amount: float,
|
|
) -> Order:
|
|
"""
|
|
Create a new order with order items.
|
|
|
|
Args:
|
|
db: Database session
|
|
obj_in: Order data
|
|
user_id: User ID
|
|
items: List of order items
|
|
total_amount: Total order amount
|
|
|
|
Returns:
|
|
The created order with items
|
|
"""
|
|
# Create order
|
|
order_data = obj_in.dict()
|
|
db_obj = Order(
|
|
**order_data,
|
|
user_id=user_id,
|
|
total_amount=total_amount,
|
|
status=OrderStatus.PENDING,
|
|
)
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
|
|
# Create order items
|
|
for item in items:
|
|
order_item = OrderItem(
|
|
order_id=db_obj.id,
|
|
product_id=item.product_id,
|
|
quantity=item.quantity,
|
|
unit_price=item.unit_price if hasattr(item, "unit_price") else 0,
|
|
)
|
|
db.add(order_item)
|
|
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def update_status(
|
|
self, db: Session, *, order_id: int, status: OrderStatus
|
|
) -> Optional[Order]:
|
|
"""
|
|
Update the status of an order.
|
|
|
|
Args:
|
|
db: Database session
|
|
order_id: Order ID
|
|
status: New order status
|
|
|
|
Returns:
|
|
The updated order or None if not found
|
|
"""
|
|
order = self.get(db, id=order_id)
|
|
if not order:
|
|
return None
|
|
|
|
order.status = status
|
|
db.add(order)
|
|
db.commit()
|
|
db.refresh(order)
|
|
return order
|
|
|
|
|
|
class CRUDOrderItem(CRUDBase[OrderItem, OrderItemCreate, OrderItemCreate]):
|
|
def get_by_order(self, db: Session, *, order_id: int) -> List[OrderItem]:
|
|
"""
|
|
Get all items for an order.
|
|
|
|
Args:
|
|
db: Database session
|
|
order_id: Order ID
|
|
|
|
Returns:
|
|
List of order items
|
|
"""
|
|
return db.query(self.model).filter(self.model.order_id == order_id).all()
|
|
|
|
|
|
order = CRUDOrder(Order)
|
|
order_item = CRUDOrderItem(OrderItem)
|