|
|
@@ -1,21 +1,198 @@
|
|
|
+"""
|
|
|
+Product Routes Module
|
|
|
+
|
|
|
+This module defines all API endpoints for product management including:
|
|
|
+- Fetching products (all users)
|
|
|
+- Creating/editing products (permissions >= 1)
|
|
|
+- Deleting products (permissions == 2 only)
|
|
|
+
|
|
|
+Permission levels:
|
|
|
+- 0: Regular user (read-only access)
|
|
|
+- 1: Manager (can create/edit products)
|
|
|
+- 2: Admin (full access including delete)
|
|
|
+"""
|
|
|
+
|
|
|
+# Standard library imports
|
|
|
from math import prod
|
|
|
+from logging import getLogger
|
|
|
+from typing import Optional
|
|
|
+
|
|
|
+# Third-party imports
|
|
|
+from fastapi import APIRouter, Depends, Query
|
|
|
from fastapi.responses import JSONResponse
|
|
|
+from h11 import Data
|
|
|
+
|
|
|
+# Local imports
|
|
|
from auth.security import get_current_user
|
|
|
-from services.data_service import ProductDataService
|
|
|
-from logging import getLogger
|
|
|
-from fastapi import APIRouter, Depends
|
|
|
-from config.messages import SuccessResponse
|
|
|
+from models import user
|
|
|
+from models.items import ProductCreateRequest, ProductEditRequest
|
|
|
+from services.data_service import DataServiceFactory
|
|
|
+from config.messages import ErrorResponse, SuccessResponse, UserResponse
|
|
|
|
|
|
+# Initialize logger for this module
|
|
|
logger = getLogger(__name__)
|
|
|
|
|
|
-product_data_service = ProductDataService()
|
|
|
+# Initialize data services for products and users
|
|
|
+product_data_service = DataServiceFactory.get_product_service()
|
|
|
+user_data_service = DataServiceFactory.get_user_service()
|
|
|
|
|
|
+# Create router instance for product-related endpoints
|
|
|
product_router = APIRouter()
|
|
|
|
|
|
@product_router.get("/")
|
|
|
-async def get_products(current_user = Depends(get_current_user)):
|
|
|
- """Get products"""
|
|
|
+async def get_products(status: Optional[int] = Query(None), current_user = Depends(get_current_user)):
|
|
|
+ """
|
|
|
+ Get all products - Available to all authenticated users
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ JSONResponse: List of all products with success message
|
|
|
+ """
|
|
|
logger.debug(f"Current user: {current_user.email if current_user else 'Anonymous'}")
|
|
|
logger.info("Fetching all products")
|
|
|
+
|
|
|
+ # Retrieve all products and convert to dictionary format
|
|
|
all_products = list(map(lambda p: p.model_dump(), product_data_service.get_all()))
|
|
|
- return JSONResponse({"products": all_products, "message": SuccessResponse.PRODUCTS_FETCH_SUCCESS})
|
|
|
+
|
|
|
+ if status is not None:
|
|
|
+ # Filter products by status if provided
|
|
|
+ all_products = [product for product in all_products if product['status'] == status]
|
|
|
+
|
|
|
+ return JSONResponse({"products": all_products, "message": SuccessResponse.PRODUCTS_FETCH_SUCCESS})
|
|
|
+
|
|
|
+@product_router.get("/{product_id}")
|
|
|
+async def get_product(product_id: int, current_user = Depends(get_current_user)):
|
|
|
+ """
|
|
|
+ Get a specific product by ID - Available to all authenticated users
|
|
|
+
|
|
|
+ Args:
|
|
|
+ product_id (int): The ID of the product to retrieve
|
|
|
+ current_user: Authenticated user (dependency injection)
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ JSONResponse: Product data if found, error message if not found
|
|
|
+ """
|
|
|
+ logger.debug(f"Current user: {current_user.email if current_user else 'Anonymous'}")
|
|
|
+ logger.info(f"Fetching product with ID: {product_id}")
|
|
|
+
|
|
|
+ # Attempt to find product by ID
|
|
|
+ product = product_data_service.get_by_id(product_id)
|
|
|
+ if product:
|
|
|
+ return JSONResponse({"product": product.model_dump(), "message": SuccessResponse.PRODUCTS_FETCH_SUCCESS})
|
|
|
+
|
|
|
+ # Return 404 if product not found
|
|
|
+ return JSONResponse({"message": UserResponse.USER_NOT_FOUND.format(user_id=product_id)}, status_code=404)
|
|
|
+
|
|
|
+# MODERATE RISK OPERATIONS - Requires permissions >= 1 (Manager level or above)
|
|
|
+
|
|
|
+@product_router.patch("/{product_id}/edit")
|
|
|
+async def edit_product(product_id: int, product: ProductEditRequest, current_user = Depends(get_current_user)):
|
|
|
+ """
|
|
|
+ Edit an existing product - Requires manager permissions (level >= 1)
|
|
|
+
|
|
|
+ Args:
|
|
|
+ product (ProductEditRequest): Product data to update
|
|
|
+ current_user: Authenticated user (dependency injection)
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ JSONResponse: Updated product data or permission denied message
|
|
|
+ """
|
|
|
+ logger.debug(f"Current user: {current_user.email if current_user else 'Anonymous'}")
|
|
|
+ logger.info(f"Editing product: {product_id}")
|
|
|
+
|
|
|
+ # Check if user has sufficient permissions (manager level or above)
|
|
|
+ if user_data_service.permissions(current_user.id) > 0:
|
|
|
+ # Update product with provided data (excluding unset fields)
|
|
|
+ product_data_service.update(product_id, **product.model_dump(exclude_unset=True))
|
|
|
+
|
|
|
+ # Retrieve updated product to return in response
|
|
|
+ edited_product = product_data_service.get_by_id(product_id)
|
|
|
+ if not edited_product:
|
|
|
+ return JSONResponse({"message": UserResponse.USER_NOT_FOUND.format(user_id=product_id)}, status_code=404)
|
|
|
+
|
|
|
+ logger.info(f"Product {product_id} edited successfully")
|
|
|
+ return JSONResponse({"message": SuccessResponse.PRODUCT_EDIT_SUCCESS, "product": edited_product.model_dump()})
|
|
|
+
|
|
|
+ # Return 403 if user lacks permissions
|
|
|
+ return JSONResponse({"message": UserResponse.NOT_PERMITTED}, status_code=403)
|
|
|
+
|
|
|
+@product_router.post("/create")
|
|
|
+async def create_product(product: ProductCreateRequest, current_user = Depends(get_current_user)):
|
|
|
+ """
|
|
|
+ Create a new product - Requires manager permissions (level >= 1)
|
|
|
+
|
|
|
+ Args:
|
|
|
+ product (ProductCreateRequest): New product data
|
|
|
+ current_user: Authenticated user (dependency injection)
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ JSONResponse: Success message or permission denied message
|
|
|
+ """
|
|
|
+ logger.debug(f"Current user: {current_user.email if current_user else 'Anonymous'}")
|
|
|
+ logger.info("Creating a new product")
|
|
|
+
|
|
|
+ # Check if user has sufficient permissions (manager level or above)
|
|
|
+ if user_data_service.permissions(current_user.id) > 0:
|
|
|
+ # Create new product with provided data
|
|
|
+ product_data_service.create(**product.model_dump(exclude_unset=True))
|
|
|
+ return JSONResponse({"message": SuccessResponse.PRODUCT_CREATE_SUCCESS, "product": product.model_dump()}, status_code=201)
|
|
|
+
|
|
|
+ # Return 403 if user lacks permissions
|
|
|
+ return JSONResponse({"message": UserResponse.NOT_PERMITTED}, status_code=403)
|
|
|
+
|
|
|
+@product_router.patch("/{product_id}/swap-status")
|
|
|
+async def switch_product_status(product_id: int, current_user = Depends(get_current_user)):
|
|
|
+ """
|
|
|
+ Toggle product status between active/inactive - Requires manager permissions (level >= 1)
|
|
|
+
|
|
|
+ Args:
|
|
|
+ product_id (int): ID of the product to update
|
|
|
+ status (int): New status value (0=inactive, 1=active)
|
|
|
+ current_user: Authenticated user (dependency injection)
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ JSONResponse: Success message or permission denied message
|
|
|
+ """
|
|
|
+ logger.debug(f"Current user: {current_user.email if current_user else 'Anonymous'}")
|
|
|
+ logger.info(f"Switching status for product with ID: {product_id}")
|
|
|
+
|
|
|
+ # Check if user has sufficient permissions (manager level or above)
|
|
|
+ if user_data_service.permissions(current_user.id) > 0:
|
|
|
+ # Update only the status field of the specified product
|
|
|
+ product = product_data_service.get_by_id(product_id)
|
|
|
+ if not product:
|
|
|
+ return JSONResponse({"message": ErrorResponse.PRODDUCT_NOT_FOUND.format(product_id=product_id)}, status_code=404)
|
|
|
+ status = 0 if product.status == 1 else 1
|
|
|
+ product_data_service.update(product_id, status=status)
|
|
|
+ return JSONResponse({"message": SuccessResponse.PRODUCT_EDIT_SUCCESS})
|
|
|
+
|
|
|
+ # Return 403 if user lacks permissions
|
|
|
+ return JSONResponse({"message": UserResponse.NOT_PERMITTED}, status_code=403)
|
|
|
+
|
|
|
+# HIGH RISK OPERATIONS - Requires permissions == 2 (Admin level only)
|
|
|
+
|
|
|
+@product_router.delete("/{product_id}")
|
|
|
+async def delete_product(product_id: int, current_user = Depends(get_current_user)):
|
|
|
+ """
|
|
|
+ Delete a product permanently - Requires admin permissions (level == 2)
|
|
|
+
|
|
|
+ This is a high-risk operation that permanently removes product data.
|
|
|
+ Only users with admin-level permissions can perform this action.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ product_id (int): ID of the product to delete
|
|
|
+ current_user: Authenticated user (dependency injection)
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ JSONResponse: Success message or permission denied message
|
|
|
+ """
|
|
|
+ logger.debug(f"Current user: {current_user.email if current_user else 'Anonymous'}")
|
|
|
+ logger.info(f"Deleting product with ID: {product_id}")
|
|
|
+
|
|
|
+ # Check if user has admin permissions (exactly level 2)
|
|
|
+ if user_data_service.permissions(current_user.id) == 2:
|
|
|
+ # Permanently delete the product
|
|
|
+ product_data_service.delete(product_id)
|
|
|
+ return JSONResponse({"message": SuccessResponse.PRODUCT_DELETE_SUCCESS})
|
|
|
+
|
|
|
+ # Return 403 if user lacks admin permissions
|
|
|
+ return JSONResponse({"message": UserResponse.NOT_PERMITTED}, status_code=403)
|
latapp