diff --git a/app/exceptions/__init__.py b/app/exceptions/__init__.py new file mode 100644 index 00000000..f41eae22 --- /dev/null +++ b/app/exceptions/__init__.py @@ -0,0 +1,120 @@ +# app/exceptions/__init__.py +""" +Custom exception classes for the LetzShop API. + +This module provides frontend-friendly exceptions with consistent error codes, +messages, and HTTP status mappings. +""" + +from .base import ( + LetzShopException, + ValidationException, + AuthenticationException, + AuthorizationException, + ResourceNotFoundException, + ConflictException, + BusinessLogicException, + ExternalServiceException, + RateLimitException, +) + +from .auth import ( + InvalidCredentialsException, + TokenExpiredException, + InsufficientPermissionsException, + UserNotActiveException, + AdminRequiredException, +) + +from .product import ( + ProductNotFoundException, + ProductAlreadyExistsException, + InvalidProductDataException, + ProductValidationException, +) + +from .stock import ( + StockNotFoundException, + InsufficientStockException, + InvalidStockOperationException, + StockValidationException, +) + +from .shop import ( + ShopNotFoundException, + ShopAlreadyExistsException, + ShopNotActiveException, + ShopNotVerifiedException, + UnauthorizedShopAccessException, + InvalidShopDataException, +) + +from .marketplace import ( + MarketplaceImportException, + ImportJobNotFoundException, + ImportJobNotOwnedException, + InvalidImportDataException, + ImportJobCannotBeCancelledException, + ImportJobCannotBeDeletedException, +) + +from .admin import ( + UserNotFoundException, + UserStatusChangeException, + ShopVerificationException, + AdminOperationException, +) + +__all__ = [ + # Base exceptions + "LetzShopException", + "ValidationException", + "AuthenticationException", + "AuthorizationException", + "ResourceNotFoundException", + "ConflictException", + "BusinessLogicException", + "ExternalServiceException", + "RateLimitException", + + # Auth exceptions + "InvalidCredentialsException", + "TokenExpiredException", + "InsufficientPermissionsException", + "UserNotActiveException", + "AdminRequiredException", + + # Product exceptions + "ProductNotFoundException", + "ProductAlreadyExistsException", + "InvalidProductDataException", + "ProductValidationException", + + # Stock exceptions + "StockNotFoundException", + "InsufficientStockException", + "InvalidStockOperationException", + "StockValidationException", + + # Shop exceptions + "ShopNotFoundException", + "ShopAlreadyExistsException", + "ShopNotActiveException", + "ShopNotVerifiedException", + "UnauthorizedShopAccessException", + "InvalidShopDataException", + + # Marketplace exceptions + "MarketplaceImportException", + "ImportJobNotFoundException", + "ImportJobNotOwnedException", + "InvalidImportDataException", + "ImportJobCannotBeCancelledException", + "ImportJobCannotBeDeletedException", + + # Admin exceptions + "UserNotFoundException", + "UserStatusChangeException", + "ShopVerificationException", + "AdminOperationException", +] diff --git a/app/exceptions/admin.py b/app/exceptions/admin.py new file mode 100644 index 00000000..4f113e15 --- /dev/null +++ b/app/exceptions/admin.py @@ -0,0 +1,191 @@ +# app/exceptions/admin.py +""" +Admin operations specific exceptions. +""" + +from typing import Any, Dict, Optional +from .base import ( + ResourceNotFoundException, + BusinessLogicException, + AuthorizationException, + ValidationException +) + + +class UserNotFoundException(ResourceNotFoundException): + """Raised when user is not found in admin operations.""" + + def __init__(self, user_identifier: str, identifier_type: str = "ID"): + if identifier_type.lower() == "username": + message = f"User with username '{user_identifier}' not found" + elif identifier_type.lower() == "email": + message = f"User with email '{user_identifier}' not found" + else: + message = f"User with ID '{user_identifier}' not found" + + super().__init__( + resource_type="User", + identifier=user_identifier, + message=message, + error_code="USER_NOT_FOUND", + ) + + +class UserStatusChangeException(BusinessLogicException): + """Raised when user status cannot be changed.""" + + def __init__( + self, + user_id: int, + current_status: str, + attempted_action: str, + reason: Optional[str] = None, + ): + message = f"Cannot {attempted_action} user {user_id} (current status: {current_status})" + if reason: + message += f": {reason}" + + super().__init__( + message=message, + error_code="USER_STATUS_CHANGE_FAILED", + details={ + "user_id": user_id, + "current_status": current_status, + "attempted_action": attempted_action, + "reason": reason, + }, + ) + + +class ShopVerificationException(BusinessLogicException): + """Raised when shop verification fails.""" + + def __init__( + self, + shop_id: int, + reason: str, + current_verification_status: Optional[bool] = None, + ): + details = { + "shop_id": shop_id, + "reason": reason, + } + + if current_verification_status is not None: + details["current_verification_status"] = current_verification_status + + super().__init__( + message=f"Shop verification failed for shop {shop_id}: {reason}", + error_code="SHOP_VERIFICATION_FAILED", + details=details, + ) + + +class AdminOperationException(BusinessLogicException): + """Raised when admin operation fails.""" + + def __init__( + self, + operation: str, + reason: str, + target_type: Optional[str] = None, + target_id: Optional[str] = None, + ): + message = f"Admin operation '{operation}' failed: {reason}" + + details = { + "operation": operation, + "reason": reason, + } + + if target_type: + details["target_type"] = target_type + if target_id: + details["target_id"] = target_id + + super().__init__( + message=message, + error_code="ADMIN_OPERATION_FAILED", + details=details, + ) + + +class CannotModifyAdminException(AuthorizationException): + """Raised when trying to modify another admin user.""" + + def __init__(self, target_user_id: int, admin_user_id: int): + super().__init__( + message=f"Cannot modify admin user {target_user_id}", + error_code="CANNOT_MODIFY_ADMIN", + details={ + "target_user_id": target_user_id, + "admin_user_id": admin_user_id, + }, + ) + + +class CannotModifySelfException(BusinessLogicException): + """Raised when admin tries to modify their own status.""" + + def __init__(self, user_id: int, operation: str): + super().__init__( + message=f"Cannot perform '{operation}' on your own account", + error_code="CANNOT_MODIFY_SELF", + details={ + "user_id": user_id, + "operation": operation, + }, + ) + + +class InvalidAdminActionException(ValidationException): + """Raised when admin action is invalid.""" + + def __init__( + self, + action: str, + reason: str, + valid_actions: Optional[list] = None, + ): + details = { + "action": action, + "reason": reason, + } + + if valid_actions: + details["valid_actions"] = valid_actions + + super().__init__( + message=f"Invalid admin action '{action}': {reason}", + details=details, + ) + self.error_code = "INVALID_ADMIN_ACTION" + + +class BulkOperationException(BusinessLogicException): + """Raised when bulk admin operation fails.""" + + def __init__( + self, + operation: str, + total_items: int, + failed_items: int, + errors: Optional[Dict[str, Any]] = None, + ): + message = f"Bulk {operation} completed with errors: {failed_items}/{total_items} failed" + + details = { + "operation": operation, + "total_items": total_items, + "failed_items": failed_items, + "success_items": total_items - failed_items, + } + + if errors: + details["errors"] = errors + + super().__init__( + message=message, + error_code="BULK_OPERATION_PARTIAL_FAILURE", + details=details, + ) diff --git a/app/exceptions/auth.py b/app/exceptions/auth.py new file mode 100644 index 00000000..fac15f78 --- /dev/null +++ b/app/exceptions/auth.py @@ -0,0 +1,95 @@ +# app/exceptions/auth.py +""" +Authentication and authorization specific exceptions. +""" + +from typing import Any, Dict, Optional +from .base import AuthenticationException, AuthorizationException + + +class InvalidCredentialsException(AuthenticationException): + """Raised when login credentials are invalid.""" + + def __init__(self, message: str = "Invalid username or password"): + super().__init__( + message=message, + error_code="INVALID_CREDENTIALS", + ) + + +class TokenExpiredException(AuthenticationException): + """Raised when JWT token has expired.""" + + def __init__(self, message: str = "Token has expired"): + super().__init__( + message=message, + error_code="TOKEN_EXPIRED", + ) + + +class InvalidTokenException(AuthenticationException): + """Raised when JWT token is invalid or malformed.""" + + def __init__(self, message: str = "Invalid token"): + super().__init__( + message=message, + error_code="INVALID_TOKEN", + ) + + +class InsufficientPermissionsException(AuthorizationException): + """Raised when user lacks required permissions for an action.""" + + def __init__( + self, + message: str = "Insufficient permissions for this action", + required_permission: Optional[str] = None, + ): + details = {} + if required_permission: + details["required_permission"] = required_permission + + super().__init__( + message=message, + error_code="INSUFFICIENT_PERMISSIONS", + details=details, + ) + + +class UserNotActiveException(AuthorizationException): + """Raised when user account is not active.""" + + def __init__(self, message: str = "User account is not active"): + super().__init__( + message=message, + error_code="USER_NOT_ACTIVE", + ) + + +class AdminRequiredException(AuthorizationException): + """Raised when admin privileges are required.""" + + def __init__(self, message: str = "Admin privileges required"): + super().__init__( + message=message, + error_code="ADMIN_REQUIRED", + ) + + +class UserAlreadyExistsException(AuthenticationException): + """Raised when trying to register with existing username/email.""" + + def __init__( + self, + message: str = "User already exists", + field: Optional[str] = None, + ): + details = {} + if field: + details["field"] = field + + super().__init__( + message=message, + error_code="USER_ALREADY_EXISTS", + details=details, + ) diff --git a/app/exceptions/base.py b/app/exceptions/base.py new file mode 100644 index 00000000..f09307ce --- /dev/null +++ b/app/exceptions/base.py @@ -0,0 +1,206 @@ +# app/exceptions/base.py +""" +Base exception classes for the LetzShop API. + +Provides consistent error structure for frontend consumption with: +- Error codes for programmatic handling +- User-friendly messages +- HTTP status code mapping +- Additional context for debugging +""" + +from typing import Any, Dict, Optional + + +class LetzShopException(Exception): + """ + Base exception for all LetzShop API errors. + + Provides consistent structure for frontend error handling. + """ + + def __init__( + self, + message: str, + error_code: str, + status_code: int = 500, + details: Optional[Dict[str, Any]] = None, + field: Optional[str] = None, + ): + self.message = message + self.error_code = error_code + self.status_code = status_code + self.details = details or {} + self.field = field + super().__init__(self.message) + + def to_dict(self) -> Dict[str, Any]: + """Convert exception to dictionary for JSON response.""" + result = { + "error_code": self.error_code, + "message": self.message, + "status_code": self.status_code, + } + + if self.field: + result["field"] = self.field + + if self.details: + result["details"] = self.details + + return result + + +class ValidationException(LetzShopException): + """Raised when input validation fails.""" + + def __init__( + self, + message: str = "Validation failed", + field: Optional[str] = None, + details: Optional[Dict[str, Any]] = None, + ): + super().__init__( + message=message, + error_code="VALIDATION_ERROR", + status_code=400, + details=details, + field=field, + ) + + +class AuthenticationException(LetzShopException): + """Raised when authentication fails.""" + + def __init__( + self, + message: str = "Authentication failed", + error_code: str = "AUTHENTICATION_FAILED", + details: Optional[Dict[str, Any]] = None, + ): + super().__init__( + message=message, + error_code=error_code, + status_code=401, + details=details, + ) + + +class AuthorizationException(LetzShopException): + """Raised when user lacks required permissions.""" + + def __init__( + self, + message: str = "Access denied", + error_code: str = "ACCESS_DENIED", + details: Optional[Dict[str, Any]] = None, + ): + super().__init__( + message=message, + error_code=error_code, + status_code=403, + details=details, + ) + + +class ResourceNotFoundException(LetzShopException): + """Raised when a requested resource is not found.""" + + def __init__( + self, + resource_type: str, + identifier: str, + message: Optional[str] = None, + error_code: Optional[str] = None, + ): + if not message: + message = f"{resource_type} not found" + + if not error_code: + error_code = f"{resource_type.upper()}_NOT_FOUND" + + super().__init__( + message=message, + error_code=error_code, + status_code=404, + details={"resource_type": resource_type, "identifier": identifier}, + ) + + +class ConflictException(LetzShopException): + """Raised when a resource conflict occurs (e.g., duplicate).""" + + def __init__( + self, + message: str, + error_code: str, + details: Optional[Dict[str, Any]] = None, + ): + super().__init__( + message=message, + error_code=error_code, + status_code=409, + details=details, + ) + + +class BusinessLogicException(LetzShopException): + """Raised when business logic validation fails.""" + + def __init__( + self, + message: str, + error_code: str, + details: Optional[Dict[str, Any]] = None, + ): + super().__init__( + message=message, + error_code=error_code, + status_code=422, + details=details, + ) + + +class ExternalServiceException(LetzShopException): + """Raised when external service calls fail.""" + + def __init__( + self, + service: str, + message: str = "External service unavailable", + error_code: str = "EXTERNAL_SERVICE_ERROR", + details: Optional[Dict[str, Any]] = None, + ): + if not details: + details = {} + details["service"] = service + + super().__init__( + message=message, + error_code=error_code, + status_code=503, + details=details, + ) + + +class RateLimitException(LetzShopException): + """Raised when rate limit is exceeded.""" + + def __init__( + self, + message: str = "Rate limit exceeded", + retry_after: Optional[int] = None, + details: Optional[Dict[str, Any]] = None, + ): + if not details: + details = {} + + if retry_after: + details["retry_after"] = retry_after + + super().__init__( + message=message, + error_code="RATE_LIMIT_EXCEEDED", + status_code=429, + details=details, + ) diff --git a/app/exceptions/handler.py b/app/exceptions/handler.py new file mode 100644 index 00000000..1ca0848a --- /dev/null +++ b/app/exceptions/handler.py @@ -0,0 +1,207 @@ +# app/exceptions/handler.py +""" +Exception handler for FastAPI application. + +Provides consistent error responses and logging for all custom exceptions. +""" + +import logging +from typing import Union + +from fastapi import Request, HTTPException +from fastapi.responses import JSONResponse +from starlette.middleware.base import BaseHTTPMiddleware + +from .base import LetzShopException + +logger = logging.getLogger(__name__) + + +class ExceptionHandlerMiddleware(BaseHTTPMiddleware): + """Middleware to handle custom exceptions and convert them to JSON responses.""" + + async def dispatch(self, request: Request, call_next): + try: + response = await call_next(request) + return response + except LetzShopException as exc: + return await self.handle_custom_exception(request, exc) + except HTTPException as exc: + return await self.handle_http_exception(request, exc) + except Exception as exc: + return await self.handle_generic_exception(request, exc) + + async def handle_custom_exception( + self, + request: Request, + exc: LetzShopException + ) -> JSONResponse: + """Handle custom LetzShop exceptions.""" + + # Log the exception + logger.error( + f"Custom exception in {request.method} {request.url}: " + f"{exc.error_code} - {exc.message}", + extra={ + "error_code": exc.error_code, + "status_code": exc.status_code, + "details": exc.details, + "url": str(request.url), + "method": request.method, + } + ) + + return JSONResponse( + status_code=exc.status_code, + content=exc.to_dict() + ) + + async def handle_http_exception( + self, + request: Request, + exc: HTTPException + ) -> JSONResponse: + """Handle FastAPI HTTPExceptions.""" + + logger.error( + f"HTTP exception in {request.method} {request.url}: " + f"{exc.status_code} - {exc.detail}", + extra={ + "status_code": exc.status_code, + "detail": exc.detail, + "url": str(request.url), + "method": request.method, + } + ) + + return JSONResponse( + status_code=exc.status_code, + content={ + "error_code": f"HTTP_{exc.status_code}", + "message": exc.detail, + "status_code": exc.status_code, + } + ) + + async def handle_generic_exception( + self, + request: Request, + exc: Exception + ) -> JSONResponse: + """Handle unexpected exceptions.""" + + logger.error( + f"Unexpected exception in {request.method} {request.url}: {str(exc)}", + exc_info=True, + extra={ + "url": str(request.url), + "method": request.method, + "exception_type": type(exc).__name__, + } + ) + + return JSONResponse( + status_code=500, + content={ + "error_code": "INTERNAL_SERVER_ERROR", + "message": "Internal server error", + "status_code": 500, + } + ) + + +def setup_exception_handlers(app): + """Setup exception handlers for the FastAPI app.""" + + @app.exception_handler(LetzShopException) + async def custom_exception_handler(request: Request, exc: LetzShopException): + """Handle custom LetzShop exceptions.""" + + logger.error( + f"Custom exception in {request.method} {request.url}: " + f"{exc.error_code} - {exc.message}", + extra={ + "error_code": exc.error_code, + "status_code": exc.status_code, + "details": exc.details, + "url": str(request.url), + "method": request.method, + } + ) + + return JSONResponse( + status_code=exc.status_code, + content=exc.to_dict() + ) + + @app.exception_handler(HTTPException) + async def http_exception_handler(request: Request, exc: HTTPException): + """Handle FastAPI HTTPExceptions with consistent format.""" + + logger.error( + f"HTTP exception in {request.method} {request.url}: " + f"{exc.status_code} - {exc.detail}", + extra={ + "status_code": exc.status_code, + "detail": exc.detail, + "url": str(request.url), + "method": request.method, + } + ) + + return JSONResponse( + status_code=exc.status_code, + content={ + "error_code": f"HTTP_{exc.status_code}", + "message": exc.detail, + "status_code": exc.status_code, + } + ) + + @app.exception_handler(Exception) + async def generic_exception_handler(request: Request, exc: Exception): + """Handle unexpected exceptions.""" + + logger.error( + f"Unexpected exception in {request.method} {request.url}: {str(exc)}", + exc_info=True, + extra={ + "url": str(request.url), + "method": request.method, + "exception_type": type(exc).__name__, + } + ) + + return JSONResponse( + status_code=500, + content={ + "error_code": "INTERNAL_SERVER_ERROR", + "message": "Internal server error", + "status_code": 500, + } + ) + + +# Utility functions for common exception scenarios +def raise_not_found(resource_type: str, identifier: str) -> None: + """Convenience function to raise ResourceNotFoundException.""" + from .base import ResourceNotFoundException + raise ResourceNotFoundException(resource_type, identifier) + + +def raise_validation_error(message: str, field: str = None, details: dict = None) -> None: + """Convenience function to raise ValidationException.""" + from .base import ValidationException + raise ValidationException(message, field, details) + + +def raise_auth_error(message: str = "Authentication failed") -> None: + """Convenience function to raise AuthenticationException.""" + from .base import AuthenticationException + raise AuthenticationException(message) + + +def raise_permission_error(message: str = "Access denied") -> None: + """Convenience function to raise AuthorizationException.""" + from .base import AuthorizationException + raise AuthorizationException(message) diff --git a/app/exceptions/marketplace.py b/app/exceptions/marketplace.py new file mode 100644 index 00000000..dd50d005 --- /dev/null +++ b/app/exceptions/marketplace.py @@ -0,0 +1,200 @@ +# app/exceptions/marketplace.py +""" +Marketplace import specific exceptions. +""" + +from typing import Any, Dict, Optional +from .base import ( + ResourceNotFoundException, + ValidationException, + BusinessLogicException, + AuthorizationException, + ExternalServiceException +) + + +class MarketplaceImportException(BusinessLogicException): + """Base exception for marketplace import operations.""" + + def __init__( + self, + message: str, + error_code: str = "MARKETPLACE_IMPORT_ERROR", + marketplace: Optional[str] = None, + details: Optional[Dict[str, Any]] = None, + ): + if not details: + details = {} + + if marketplace: + details["marketplace"] = marketplace + + super().__init__( + message=message, + error_code=error_code, + details=details, + ) + + +class ImportJobNotFoundException(ResourceNotFoundException): + """Raised when import job is not found.""" + + def __init__(self, job_id: int): + super().__init__( + resource_type="ImportJob", + identifier=str(job_id), + message=f"Import job with ID '{job_id}' not found", + error_code="IMPORT_JOB_NOT_FOUND", + ) + + +class ImportJobNotOwnedException(AuthorizationException): + """Raised when user tries to access import job they don't own.""" + + def __init__(self, job_id: int, user_id: Optional[int] = None): + details = {"job_id": job_id} + if user_id: + details["user_id"] = user_id + + super().__init__( + message=f"Unauthorized access to import job '{job_id}'", + error_code="IMPORT_JOB_NOT_OWNED", + details=details, + ) + + +class InvalidImportDataException(ValidationException): + """Raised when import data is invalid.""" + + def __init__( + self, + message: str = "Invalid import data", + field: Optional[str] = None, + row_number: Optional[int] = None, + details: Optional[Dict[str, Any]] = None, + ): + if not details: + details = {} + + if row_number: + details["row_number"] = row_number + + super().__init__( + message=message, + field=field, + details=details, + ) + self.error_code = "INVALID_IMPORT_DATA" + + +class ImportJobCannotBeCancelledException(BusinessLogicException): + """Raised when trying to cancel job that cannot be cancelled.""" + + def __init__(self, job_id: int, current_status: str): + super().__init__( + message=f"Import job '{job_id}' cannot be cancelled (current status: {current_status})", + error_code="IMPORT_JOB_CANNOT_BE_CANCELLED", + details={ + "job_id": job_id, + "current_status": current_status, + }, + ) + + +class ImportJobCannotBeDeletedException(BusinessLogicException): + """Raised when trying to delete job that cannot be deleted.""" + + def __init__(self, job_id: int, current_status: str): + super().__init__( + message=f"Import job '{job_id}' cannot be deleted (current status: {current_status})", + error_code="IMPORT_JOB_CANNOT_BE_DELETED", + details={ + "job_id": job_id, + "current_status": current_status, + }, + ) + + +class MarketplaceConnectionException(ExternalServiceException): + """Raised when marketplace connection fails.""" + + def __init__(self, marketplace: str, message: str = "Failed to connect to marketplace"): + super().__init__( + service=marketplace, + message=f"{message}: {marketplace}", + error_code="MARKETPLACE_CONNECTION_FAILED", + ) + + +class MarketplaceDataParsingException(ValidationException): + """Raised when marketplace data cannot be parsed.""" + + def __init__( + self, + marketplace: str, + message: str = "Failed to parse marketplace data", + details: Optional[Dict[str, Any]] = None, + ): + if not details: + details = {} + details["marketplace"] = marketplace + + super().__init__( + message=f"{message} from {marketplace}", + details=details, + ) + self.error_code = "MARKETPLACE_DATA_PARSING_FAILED" + + +class ImportRateLimitException(BusinessLogicException): + """Raised when import rate limit is exceeded.""" + + def __init__( + self, + max_imports: int, + time_window: str, + retry_after: Optional[int] = None, + ): + details = { + "max_imports": max_imports, + "time_window": time_window, + } + + if retry_after: + details["retry_after"] = retry_after + + super().__init__( + message=f"Import rate limit exceeded: {max_imports} imports per {time_window}", + error_code="IMPORT_RATE_LIMIT_EXCEEDED", + details=details, + ) + + +class InvalidMarketplaceException(ValidationException): + """Raised when marketplace is not supported.""" + + def __init__(self, marketplace: str, supported_marketplaces: Optional[list] = None): + details = {"marketplace": marketplace} + if supported_marketplaces: + details["supported_marketplaces"] = supported_marketplaces + + super().__init__( + message=f"Unsupported marketplace: {marketplace}", + field="marketplace", + details=details, + ) + self.error_code = "INVALID_MARKETPLACE" + + +class ImportJobAlreadyProcessingException(BusinessLogicException): + """Raised when trying to start import while another is already processing.""" + + def __init__(self, shop_code: str, existing_job_id: int): + super().__init__( + message=f"Import already in progress for shop '{shop_code}'", + error_code="IMPORT_JOB_ALREADY_PROCESSING", + details={ + "shop_code": shop_code, + "existing_job_id": existing_job_id, + }, + ) diff --git a/app/exceptions/product.py b/app/exceptions/product.py new file mode 100644 index 00000000..88b77585 --- /dev/null +++ b/app/exceptions/product.py @@ -0,0 +1,102 @@ +# app/exceptions/product.py +""" +Product management specific exceptions. +""" + +from typing import Any, Dict, Optional +from .base import ResourceNotFoundException, ConflictException, ValidationException, BusinessLogicException + + +class ProductNotFoundException(ResourceNotFoundException): + """Raised when a product is not found.""" + + def __init__(self, product_id: str): + super().__init__( + resource_type="Product", + identifier=product_id, + message=f"Product with ID '{product_id}' not found", + error_code="PRODUCT_NOT_FOUND", + ) + + +class ProductAlreadyExistsException(ConflictException): + """Raised when trying to create a product that already exists.""" + + def __init__(self, product_id: str): + super().__init__( + message=f"Product with ID '{product_id}' already exists", + error_code="PRODUCT_ALREADY_EXISTS", + details={"product_id": product_id}, + ) + + +class InvalidProductDataException(ValidationException): + """Raised when product data is invalid.""" + + def __init__( + self, + message: str = "Invalid product data", + field: Optional[str] = None, + details: Optional[Dict[str, Any]] = None, + ): + super().__init__( + message=message, + field=field, + details=details, + ) + self.error_code = "INVALID_PRODUCT_DATA" + + +class ProductValidationException(ValidationException): + """Raised when product validation fails.""" + + def __init__( + self, + message: str, + field: Optional[str] = None, + validation_errors: Optional[Dict[str, str]] = None, + ): + details = {} + if validation_errors: + details["validation_errors"] = validation_errors + + super().__init__( + message=message, + field=field, + details=details, + ) + self.error_code = "PRODUCT_VALIDATION_FAILED" + + +class InvalidGTINException(ValidationException): + """Raised when GTIN format is invalid.""" + + def __init__(self, gtin: str, message: str = "Invalid GTIN format"): + super().__init__( + message=f"{message}: {gtin}", + field="gtin", + details={"gtin": gtin}, + ) + self.error_code = "INVALID_GTIN" + + +class ProductCSVImportException(BusinessLogicException): + """Raised when product CSV import fails.""" + + def __init__( + self, + message: str = "Product CSV import failed", + row_number: Optional[int] = None, + errors: Optional[Dict[str, Any]] = None, + ): + details = {} + if row_number: + details["row_number"] = row_number + if errors: + details["errors"] = errors + + super().__init__( + message=message, + error_code="PRODUCT_CSV_IMPORT_FAILED", + details=details, + ) diff --git a/app/exceptions/shop.py b/app/exceptions/shop.py new file mode 100644 index 00000000..95aeb613 --- /dev/null +++ b/app/exceptions/shop.py @@ -0,0 +1,157 @@ +# app/exceptions/shop.py +""" +Shop management specific exceptions. +""" + +from typing import Any, Dict, Optional +from .base import ( + ResourceNotFoundException, + ConflictException, + ValidationException, + AuthorizationException, + BusinessLogicException +) + + +class ShopNotFoundException(ResourceNotFoundException): + """Raised when a shop is not found.""" + + def __init__(self, shop_identifier: str, identifier_type: str = "code"): + if identifier_type.lower() == "id": + message = f"Shop with ID '{shop_identifier}' not found" + else: + message = f"Shop with code '{shop_identifier}' not found" + + super().__init__( + resource_type="Shop", + identifier=shop_identifier, + message=message, + error_code="SHOP_NOT_FOUND", + ) + + +class ShopAlreadyExistsException(ConflictException): + """Raised when trying to create a shop that already exists.""" + + def __init__(self, shop_code: str): + super().__init__( + message=f"Shop with code '{shop_code}' already exists", + error_code="SHOP_ALREADY_EXISTS", + details={"shop_code": shop_code}, + ) + + +class ShopNotActiveException(BusinessLogicException): + """Raised when trying to perform operations on inactive shop.""" + + def __init__(self, shop_code: str): + super().__init__( + message=f"Shop '{shop_code}' is not active", + error_code="SHOP_NOT_ACTIVE", + details={"shop_code": shop_code}, + ) + + +class ShopNotVerifiedException(BusinessLogicException): + """Raised when trying to perform operations requiring verified shop.""" + + def __init__(self, shop_code: str): + super().__init__( + message=f"Shop '{shop_code}' is not verified", + error_code="SHOP_NOT_VERIFIED", + details={"shop_code": shop_code}, + ) + + +class UnauthorizedShopAccessException(AuthorizationException): + """Raised when user tries to access shop they don't own.""" + + def __init__(self, shop_code: str, user_id: Optional[int] = None): + details = {"shop_code": shop_code} + if user_id: + details["user_id"] = user_id + + super().__init__( + message=f"Unauthorized access to shop '{shop_code}'", + error_code="UNAUTHORIZED_SHOP_ACCESS", + details=details, + ) + + +class InvalidShopDataException(ValidationException): + """Raised when shop data is invalid.""" + + def __init__( + self, + message: str = "Invalid shop data", + field: Optional[str] = None, + details: Optional[Dict[str, Any]] = None, + ): + super().__init__( + message=message, + field=field, + details=details, + ) + self.error_code = "INVALID_SHOP_DATA" + + +class ShopProductAlreadyExistsException(ConflictException): + """Raised when trying to add a product that already exists in shop.""" + + def __init__(self, shop_code: str, product_id: str): + super().__init__( + message=f"Product '{product_id}' already exists in shop '{shop_code}'", + error_code="SHOP_PRODUCT_ALREADY_EXISTS", + details={ + "shop_code": shop_code, + "product_id": product_id, + }, + ) + + +class ShopProductNotFoundException(ResourceNotFoundException): + """Raised when a shop product relationship is not found.""" + + def __init__(self, shop_code: str, product_id: str): + super().__init__( + resource_type="ShopProduct", + identifier=f"{shop_code}/{product_id}", + message=f"Product '{product_id}' not found in shop '{shop_code}'", + error_code="SHOP_PRODUCT_NOT_FOUND", + ) + + +class MaxShopsReachedException(BusinessLogicException): + """Raised when user tries to create more shops than allowed.""" + + def __init__(self, max_shops: int, user_id: Optional[int] = None): + details = {"max_shops": max_shops} + if user_id: + details["user_id"] = user_id + + super().__init__( + message=f"Maximum number of shops reached ({max_shops})", + error_code="MAX_SHOPS_REACHED", + details=details, + ) + + +class ShopValidationException(ValidationException): + """Raised when shop validation fails.""" + + def __init__( + self, + message: str = "Shop validation failed", + field: Optional[str] = None, + validation_errors: Optional[Dict[str, str]] = None, + ): + details = {} + if validation_errors: + details["validation_errors"] = validation_errors + + super().__init__( + message=message, + field=field, + details=details, + ) + self.error_code = "SHOP_VALIDATION_FAILED" diff --git a/app/exceptions/stock.py b/app/exceptions/stock.py new file mode 100644 index 00000000..cad0eab6 --- /dev/null +++ b/app/exceptions/stock.py @@ -0,0 +1,132 @@ +# app/exceptions/stock.py +""" +Stock management specific exceptions. +""" + +from typing import Any, Dict, Optional +from .base import ResourceNotFoundException, ValidationException, BusinessLogicException + + +class StockNotFoundException(ResourceNotFoundException): + """Raised when stock record is not found.""" + + def __init__(self, identifier: str, identifier_type: str = "ID"): + if identifier_type.lower() == "gtin": + message = f"No stock found for GTIN '{identifier}'" + else: + message = f"Stock record with {identifier_type} '{identifier}' not found" + + super().__init__( + resource_type="Stock", + identifier=identifier, + message=message, + error_code="STOCK_NOT_FOUND", + ) + + +class InsufficientStockException(BusinessLogicException): + """Raised when trying to remove more stock than available.""" + + def __init__( + self, + gtin: str, + location: str, + requested: int, + available: int, + ): + message = f"Insufficient stock for GTIN '{gtin}' at '{location}'. Requested: {requested}, Available: {available}" + + super().__init__( + message=message, + error_code="INSUFFICIENT_STOCK", + details={ + "gtin": gtin, + "location": location, + "requested_quantity": requested, + "available_quantity": available, + }, + ) + + +class InvalidStockOperationException(ValidationException): + """Raised when stock operation is invalid.""" + + def __init__( + self, + message: str, + operation: Optional[str] = None, + details: Optional[Dict[str, Any]] = None, + ): + if not details: + details = {} + + if operation: + details["operation"] = operation + + super().__init__( + message=message, + details=details, + ) + self.error_code = "INVALID_STOCK_OPERATION" + + +class StockValidationException(ValidationException): + """Raised when stock data validation fails.""" + + def __init__( + self, + message: str = "Stock validation failed", + field: Optional[str] = None, + validation_errors: Optional[Dict[str, str]] = None, + ): + details = {} + if validation_errors: + details["validation_errors"] = validation_errors + + super().__init__( + message=message, + field=field, + details=details, + ) + self.error_code = "STOCK_VALIDATION_FAILED" + + +class NegativeStockException(BusinessLogicException): + """Raised when stock quantity would become negative.""" + + def __init__(self, gtin: str, location: str, resulting_quantity: int): + message = f"Stock operation would result in negative quantity ({resulting_quantity}) for GTIN '{gtin}' at '{location}'" + + super().__init__( + message=message, + error_code="NEGATIVE_STOCK_NOT_ALLOWED", + details={ + "gtin": gtin, + "location": location, + "resulting_quantity": resulting_quantity, + }, + ) + + +class InvalidQuantityException(ValidationException): + """Raised when quantity value is invalid.""" + + def __init__(self, quantity: Any, message: str = "Invalid quantity"): + super().__init__( + message=f"{message}: {quantity}", + field="quantity", + details={"quantity": quantity}, + ) + self.error_code = "INVALID_QUANTITY" + + +class LocationNotFoundException(ResourceNotFoundException): + """Raised when stock location is not found.""" + + def __init__(self, location: str): + super().__init__( + resource_type="Location", + identifier=location, + message=f"Stock location '{location}' not found", + error_code="LOCATION_NOT_FOUND", + )