Samir Boulahtit
1b8a40f1ff
All checks were successful
- Add centralized _is_noqa_suppressed() to BaseValidator with normalization (accepts both SEC001 and SEC-001 formats for ruff compatibility) - Wire noqa support into all 21 security and 18 performance check functions - Add ruff external config for SEC/PERF/MOD/EXC codes in pyproject.toml - Convert all 280 Python noqa comments to dashless format (ruff-compatible) - Add site/ to IGNORE_PATTERNS (excludes mkdocs build output) - Suppress 152 false positive findings (test passwords, seed data, validator self-references, Apple Wallet SHA1, etc.) - Security: 79 errors → 0, 60 warnings → 0 - Performance: 80 warnings → 77 (3 test script suppressions) - Add proposal doc with noqa inventory and remaining findings recommendations Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
596 lines
18 KiB
Python
596 lines
18 KiB
Python
# app/modules/marketplace/exceptions.py
|
|
"""
|
|
Marketplace module exceptions.
|
|
|
|
This module provides exception classes for marketplace operations including:
|
|
- Letzshop integration (client, authentication, credentials)
|
|
- Product import/export
|
|
- Sync operations
|
|
- Onboarding wizard
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
from app.exceptions.base import (
|
|
AuthorizationException,
|
|
BusinessLogicException,
|
|
ConflictException,
|
|
ExternalServiceException,
|
|
ResourceNotFoundException,
|
|
ValidationException,
|
|
)
|
|
|
|
__all__ = [
|
|
# Base exception
|
|
"MarketplaceException",
|
|
# Letzshop client exceptions
|
|
"LetzshopClientError",
|
|
"LetzshopAuthenticationError",
|
|
"LetzshopCredentialsNotFoundException",
|
|
"LetzshopConnectionFailedException",
|
|
# Import job exceptions
|
|
"ImportJobNotFoundException",
|
|
"HistoricalImportJobNotFoundException",
|
|
"ImportJobNotOwnedException",
|
|
"ImportJobCannotBeCancelledException",
|
|
"ImportJobCannotBeDeletedException",
|
|
"ImportJobAlreadyProcessingException",
|
|
"ImportValidationError",
|
|
"InvalidImportDataException",
|
|
"ImportRateLimitException",
|
|
# Marketplace exceptions
|
|
"MarketplaceImportException",
|
|
"MarketplaceConnectionException",
|
|
"MarketplaceDataParsingException",
|
|
"InvalidMarketplaceException",
|
|
# Product exceptions
|
|
"StoreNotFoundException",
|
|
"ProductNotFoundException",
|
|
"MarketplaceProductNotFoundException",
|
|
"MarketplaceProductAlreadyExistsException",
|
|
"InvalidMarketplaceProductDataException",
|
|
"MarketplaceProductValidationException",
|
|
"InvalidGTINException",
|
|
"MarketplaceProductCSVImportException",
|
|
# Export/Sync exceptions
|
|
"ExportError",
|
|
"SyncError",
|
|
# Onboarding exceptions
|
|
"OnboardingNotFoundException",
|
|
"OnboardingStepOrderException",
|
|
"OnboardingAlreadyCompletedException",
|
|
"OnboardingCsvUrlRequiredException",
|
|
"OnboardingSyncJobNotFoundException",
|
|
"OnboardingSyncNotCompleteException",
|
|
]
|
|
|
|
|
|
# =============================================================================
|
|
# Base Marketplace Exception
|
|
# =============================================================================
|
|
|
|
|
|
class MarketplaceException(BusinessLogicException):
|
|
"""Base exception for marketplace module errors."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
error_code: str = "MARKETPLACE_ERROR",
|
|
details: dict | None = None,
|
|
):
|
|
super().__init__(message=message, error_code=error_code, details=details)
|
|
|
|
|
|
# =============================================================================
|
|
# Letzshop Client Exceptions
|
|
# =============================================================================
|
|
|
|
|
|
class LetzshopClientError(MarketplaceException):
|
|
"""Raised when Letzshop API call fails."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
status_code: int | None = None,
|
|
response: str | None = None,
|
|
):
|
|
details = {}
|
|
if status_code:
|
|
details["http_status_code"] = status_code
|
|
if response:
|
|
details["response"] = response
|
|
super().__init__(
|
|
message=message,
|
|
error_code="LETZSHOP_CLIENT_ERROR",
|
|
details=details if details else None,
|
|
)
|
|
self.status_code = status_code
|
|
self.response = response
|
|
|
|
|
|
class LetzshopAuthenticationError(LetzshopClientError): # noqa: MOD025
|
|
"""Raised when Letzshop authentication fails."""
|
|
|
|
def __init__(self, message: str = "Letzshop authentication failed"):
|
|
super().__init__(message, status_code=401)
|
|
self.error_code = "LETZSHOP_AUTHENTICATION_FAILED"
|
|
|
|
|
|
class LetzshopCredentialsNotFoundException(ResourceNotFoundException): # noqa: MOD025
|
|
"""Raised when Letzshop credentials not found for store."""
|
|
|
|
def __init__(self, store_id: int):
|
|
super().__init__(
|
|
resource_type="LetzshopCredentials",
|
|
identifier=str(store_id),
|
|
error_code="LETZSHOP_CREDENTIALS_NOT_FOUND",
|
|
)
|
|
self.store_id = store_id
|
|
|
|
|
|
class LetzshopConnectionFailedException(BusinessLogicException): # noqa: MOD025
|
|
"""Raised when Letzshop API connection test fails."""
|
|
|
|
def __init__(self, error_message: str):
|
|
super().__init__(
|
|
message=f"Letzshop connection failed: {error_message}",
|
|
error_code="LETZSHOP_CONNECTION_FAILED",
|
|
details={"error": error_message},
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Import Job Exceptions
|
|
# =============================================================================
|
|
|
|
|
|
class ImportJobNotFoundException(ResourceNotFoundException):
|
|
"""Raised when a marketplace import job is not found."""
|
|
|
|
def __init__(self, job_id: int):
|
|
super().__init__(
|
|
resource_type="MarketplaceImportJob",
|
|
identifier=str(job_id),
|
|
message=f"Import job with ID '{job_id}' not found",
|
|
error_code="IMPORT_JOB_NOT_FOUND",
|
|
)
|
|
|
|
|
|
class HistoricalImportJobNotFoundException(ResourceNotFoundException): # noqa: MOD025
|
|
"""Raised when a historical import job is not found."""
|
|
|
|
def __init__(self, job_id: int):
|
|
super().__init__(
|
|
resource_type="LetzshopHistoricalImportJob",
|
|
identifier=str(job_id),
|
|
error_code="HISTORICAL_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: int | None = 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 ImportJobCannotBeCancelledException(BusinessLogicException): # noqa: MOD025
|
|
"""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): # noqa: MOD025
|
|
"""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 ImportJobAlreadyProcessingException(BusinessLogicException): # noqa: MOD025
|
|
"""Raised when trying to start import while another is already processing."""
|
|
|
|
def __init__(self, store_code: str, existing_job_id: int):
|
|
super().__init__(
|
|
message=f"Import already in progress for store '{store_code}'",
|
|
error_code="IMPORT_JOB_ALREADY_PROCESSING",
|
|
details={
|
|
"store_code": store_code,
|
|
"existing_job_id": existing_job_id,
|
|
},
|
|
)
|
|
|
|
|
|
class ImportValidationError(MarketplaceException):
|
|
"""Raised when import data validation fails."""
|
|
|
|
def __init__(self, message: str, errors: list[dict] | None = None):
|
|
super().__init__(
|
|
message=message,
|
|
error_code="IMPORT_VALIDATION_ERROR",
|
|
details={"errors": errors} if errors else None,
|
|
)
|
|
self.errors = errors or []
|
|
|
|
|
|
class InvalidImportDataException(ValidationException): # noqa: MOD025
|
|
"""Raised when import data is invalid."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Invalid import data",
|
|
field: str | None = None,
|
|
row_number: int | None = None,
|
|
details: dict[str, Any] | None = 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 ImportRateLimitException(BusinessLogicException): # noqa: MOD025
|
|
"""Raised when import rate limit is exceeded."""
|
|
|
|
def __init__(
|
|
self,
|
|
max_imports: int,
|
|
time_window: str,
|
|
retry_after: int | None = 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,
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Marketplace Exceptions
|
|
# =============================================================================
|
|
|
|
|
|
class MarketplaceImportException(BusinessLogicException): # noqa: MOD025
|
|
"""Base exception for marketplace import operations."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
error_code: str = "MARKETPLACE_IMPORT_ERROR",
|
|
marketplace: str | None = None,
|
|
details: dict[str, Any] | None = None,
|
|
):
|
|
if not details:
|
|
details = {}
|
|
|
|
if marketplace:
|
|
details["marketplace"] = marketplace
|
|
|
|
super().__init__(
|
|
message=message,
|
|
error_code=error_code,
|
|
details=details,
|
|
)
|
|
|
|
|
|
class MarketplaceConnectionException(ExternalServiceException): # noqa: MOD025
|
|
"""Raised when marketplace connection fails."""
|
|
|
|
def __init__(
|
|
self, marketplace: str, message: str = "Failed to connect to marketplace"
|
|
):
|
|
super().__init__(
|
|
service_name=marketplace,
|
|
message=f"{message}: {marketplace}",
|
|
error_code="MARKETPLACE_CONNECTION_FAILED",
|
|
)
|
|
|
|
|
|
class MarketplaceDataParsingException(ValidationException): # noqa: MOD025
|
|
"""Raised when marketplace data cannot be parsed."""
|
|
|
|
def __init__(
|
|
self,
|
|
marketplace: str,
|
|
message: str = "Failed to parse marketplace data",
|
|
details: dict[str, Any] | None = 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 InvalidMarketplaceException(ValidationException): # noqa: MOD025
|
|
"""Raised when marketplace is not supported."""
|
|
|
|
def __init__(self, marketplace: str, supported_marketplaces: list | None = 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"
|
|
|
|
|
|
# =============================================================================
|
|
# Product Exceptions
|
|
# =============================================================================
|
|
|
|
|
|
class StoreNotFoundException(ResourceNotFoundException):
|
|
"""Raised when a store is not found."""
|
|
|
|
def __init__(self, store_id: int):
|
|
super().__init__(
|
|
resource_type="Store",
|
|
identifier=str(store_id),
|
|
error_code="STORE_NOT_FOUND",
|
|
)
|
|
|
|
|
|
class ProductNotFoundException(ResourceNotFoundException):
|
|
"""Raised when a marketplace product is not found."""
|
|
|
|
def __init__(self, product_id: str | int):
|
|
super().__init__(
|
|
resource_type="MarketplaceProduct",
|
|
identifier=str(product_id),
|
|
error_code="MARKETPLACE_PRODUCT_NOT_FOUND",
|
|
)
|
|
|
|
|
|
class MarketplaceProductNotFoundException(ResourceNotFoundException):
|
|
"""Raised when a product is not found."""
|
|
|
|
def __init__(self, marketplace_product_id: str):
|
|
super().__init__(
|
|
resource_type="MarketplaceProduct",
|
|
identifier=marketplace_product_id,
|
|
message=f"MarketplaceProduct with ID '{marketplace_product_id}' not found",
|
|
error_code="PRODUCT_NOT_FOUND",
|
|
)
|
|
|
|
|
|
class MarketplaceProductAlreadyExistsException(ConflictException):
|
|
"""Raised when trying to create a product that already exists."""
|
|
|
|
def __init__(self, marketplace_product_id: str):
|
|
super().__init__(
|
|
message=f"MarketplaceProduct with ID '{marketplace_product_id}' already exists",
|
|
error_code="PRODUCT_ALREADY_EXISTS",
|
|
details={"marketplace_product_id": marketplace_product_id},
|
|
)
|
|
|
|
|
|
class InvalidMarketplaceProductDataException(ValidationException):
|
|
"""Raised when product data is invalid."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "Invalid product data",
|
|
field: str | None = None,
|
|
details: dict[str, Any] | None = None,
|
|
):
|
|
super().__init__(
|
|
message=message,
|
|
field=field,
|
|
details=details,
|
|
)
|
|
self.error_code = "INVALID_PRODUCT_DATA"
|
|
|
|
|
|
class MarketplaceProductValidationException(ValidationException):
|
|
"""Raised when product validation fails."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
field: str | None = None,
|
|
validation_errors: dict[str, str] | None = 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): # noqa: MOD025
|
|
"""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 MarketplaceProductCSVImportException(BusinessLogicException): # noqa: MOD025
|
|
"""Raised when product CSV import fails."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "MarketplaceProduct CSV import failed",
|
|
row_number: int | None = None,
|
|
errors: dict[str, Any] | None = 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,
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Export/Sync Exceptions
|
|
# =============================================================================
|
|
|
|
|
|
class ExportError(MarketplaceException): # noqa: MOD025
|
|
"""Raised when product export fails."""
|
|
|
|
def __init__(self, message: str, language: str | None = None):
|
|
details = {}
|
|
if language:
|
|
details["language"] = language
|
|
super().__init__(
|
|
message=message,
|
|
error_code="EXPORT_ERROR",
|
|
details=details if details else None,
|
|
)
|
|
self.language = language
|
|
|
|
|
|
class SyncError(MarketplaceException):
|
|
"""Raised when store directory sync fails."""
|
|
|
|
def __init__(self, message: str, store_code: str | None = None):
|
|
details = {}
|
|
if store_code:
|
|
details["store_code"] = store_code
|
|
super().__init__(
|
|
message=message,
|
|
error_code="SYNC_ERROR",
|
|
details=details if details else None,
|
|
)
|
|
self.store_code = store_code
|
|
|
|
|
|
# =============================================================================
|
|
# Onboarding Exceptions
|
|
# =============================================================================
|
|
|
|
|
|
class OnboardingNotFoundException(ResourceNotFoundException):
|
|
"""Raised when onboarding record is not found for a store."""
|
|
|
|
def __init__(self, store_id: int):
|
|
super().__init__(
|
|
resource_type="StoreOnboarding",
|
|
identifier=str(store_id),
|
|
error_code="ONBOARDING_NOT_FOUND",
|
|
)
|
|
|
|
|
|
class OnboardingStepOrderException(ValidationException):
|
|
"""Raised when trying to access a step out of order."""
|
|
|
|
def __init__(self, current_step: str, required_step: str):
|
|
super().__init__(
|
|
message=f"Please complete the {required_step} step first",
|
|
field="step",
|
|
details={
|
|
"current_step": current_step,
|
|
"required_step": required_step,
|
|
},
|
|
)
|
|
self.error_code = "ONBOARDING_STEP_ORDER_ERROR"
|
|
|
|
|
|
class OnboardingAlreadyCompletedException(BusinessLogicException):
|
|
"""Raised when trying to modify a completed onboarding."""
|
|
|
|
def __init__(self, store_id: int):
|
|
super().__init__(
|
|
message="Onboarding has already been completed",
|
|
error_code="ONBOARDING_ALREADY_COMPLETED",
|
|
details={"store_id": store_id},
|
|
)
|
|
|
|
|
|
class OnboardingCsvUrlRequiredException(ValidationException):
|
|
"""Raised when no CSV URL is provided in product import step."""
|
|
|
|
def __init__(self):
|
|
super().__init__(
|
|
message="At least one CSV URL must be provided",
|
|
field="csv_url",
|
|
)
|
|
self.error_code = "ONBOARDING_CSV_URL_REQUIRED"
|
|
|
|
|
|
class OnboardingSyncJobNotFoundException(ResourceNotFoundException):
|
|
"""Raised when sync job is not found."""
|
|
|
|
def __init__(self, job_id: int):
|
|
super().__init__(
|
|
resource_type="LetzshopHistoricalImportJob",
|
|
identifier=str(job_id),
|
|
error_code="ONBOARDING_SYNC_JOB_NOT_FOUND",
|
|
)
|
|
|
|
|
|
class OnboardingSyncNotCompleteException(BusinessLogicException):
|
|
"""Raised when trying to complete onboarding before sync is done."""
|
|
|
|
def __init__(self, job_status: str):
|
|
super().__init__(
|
|
message=f"Import job is still {job_status}, please wait",
|
|
error_code="SYNC_NOT_COMPLETE",
|
|
details={"job_status": job_status},
|
|
)
|