Upload core/error_enforcement.py with huggingface_hub

core/error_enforcement.py

@@ -71,7 +71,9 @@ class ErrorResponseMiddleware(BaseHTTPMiddleware):
         except (json.JSONDecodeError, AttributeError, KeyError):
             return False
 
-    def _handle_http_exception(self, request: Request, exc: HTTPException) -> JSONResponse:
+    def _handle_http_exception(
+        self, request: Request, exc: HTTPException
+    ) -> JSONResponse:
         """Handle HTTPException with standardized response"""
         # Determine error type based on status code
         error_type = self._get_error_type_from_status(exc.status_code)
@@ -97,9 +99,15 @@ class ErrorResponseMiddleware(BaseHTTPMiddleware):
             request=request,
         )
 
-        return JSONResponse(status_code=exc.status_code, content=error_response, headers=self._get_security_headers())
+        return JSONResponse(
+            status_code=exc.status_code,
+            content=error_response,
+            headers=self._get_security_headers(),
+        )
 
-    def _handle_starlette_exception(self, request: Request, exc: StarletteHTTPException) -> JSONResponse:
+    def _handle_starlette_exception(
+        self, request: Request, exc: StarletteHTTPException
+    ) -> JSONResponse:
         """Handle Starlette HTTPException with standardized response"""
         error_type = self._get_error_type_from_status(exc.status_code)
 
@@ -124,9 +132,15 @@ class ErrorResponseMiddleware(BaseHTTPMiddleware):
             request=request,
         )
 
-        return JSONResponse(status_code=exc.status_code, content=error_response, headers=self._get_security_headers())
+        return JSONResponse(
+            status_code=exc.status_code,
+            content=error_response,
+            headers=self._get_security_headers(),
+        )
 
-    def _handle_unexpected_exception(self, request: Request, exc: Exception) -> JSONResponse:
+    def _handle_unexpected_exception(
+        self, request: Request, exc: Exception
+    ) -> JSONResponse:
         """Handle unexpected exceptions with standardized response"""
         # Log the full error
         error_details = {
@@ -166,7 +180,11 @@ class ErrorResponseMiddleware(BaseHTTPMiddleware):
                 request=request,
             )
 
-        return JSONResponse(status_code=500, content=error_response, headers=self._get_security_headers())
+        return JSONResponse(
+            status_code=500,
+            content=error_response,
+            headers=self._get_security_headers(),
+        )
 
     def _get_error_type_from_status(self, status_code: int) -> str:
         """Map HTTP status codes to error types"""
@@ -221,7 +239,9 @@ class ValidationErrorMiddleware(BaseHTTPMiddleware):
 
         return isinstance(exc, (ValidationError, RequestValidationError))
 
-    def _handle_validation_error(self, request: Request, exc: Exception) -> JSONResponse:
+    def _handle_validation_error(
+        self, request: Request, exc: Exception
+    ) -> JSONResponse:
         """Handle validation errors with detailed response"""
         from fastapi.exceptions import RequestValidationError
         from pydantic import ValidationError
@@ -275,7 +295,11 @@ class ValidationErrorMiddleware(BaseHTTPMiddleware):
             details=validation_details,
         )
 
-        return JSONResponse(status_code=422, content=error_response, headers=self._get_security_headers())
+        return JSONResponse(
+            status_code=422,
+            content=error_response,
+            headers=self._get_security_headers(),
+        )
 
     def _get_security_headers(self) -> Dict[str, str]:
         """Get security headers for validation error responses"""
@@ -298,14 +322,34 @@ class BusinessErrorMiddleware(BaseHTTPMiddleware):
         super().__init__(app)
         self._business_error_mapping = {
             "UserNotFoundError": (404, "user_not_found", "User not found"),
-            "InvalidCredentialsError": (401, "invalid_credentials", "Invalid credentials"),
+            "InvalidCredentialsError": (
+                401,
+                "invalid_credentials",
+                "Invalid credentials",
+            ),
             "AccountLockedError": (423, "account_locked", "Account is locked"),
-            "InsufficientPermissionsError": (403, "insufficient_permissions", "Insufficient permissions"),
+            "InsufficientPermissionsError": (
+                403,
+                "insufficient_permissions",
+                "Insufficient permissions",
+            ),
             "ResourceNotFoundError": (404, "resource_not_found", "Resource not found"),
             "ResourceConflictError": (409, "resource_conflict", "Resource conflict"),
-            "BusinessRuleViolationError": (422, "business_rule_violation", "Business rule violation"),
-            "RateLimitExceededError": (429, "rate_limit_exceeded", "Rate limit exceeded"),
-            "ServiceUnavailableError": (503, "service_unavailable", "Service temporarily unavailable"),
+            "BusinessRuleViolationError": (
+                422,
+                "business_rule_violation",
+                "Business rule violation",
+            ),
+            "RateLimitExceededError": (
+                429,
+                "rate_limit_exceeded",
+                "Rate limit exceeded",
+            ),
+            "ServiceUnavailableError": (
+                503,
+                "service_unavailable",
+                "Service temporarily unavailable",
+            ),
         }
 
     async def dispatch(self, request: Request, call_next):
@@ -320,9 +364,13 @@ class BusinessErrorMiddleware(BaseHTTPMiddleware):
             # Let other errors be handled by other middleware
             raise
 
-    def _handle_business_error(self, request: Request, exc: Exception, error_name: str) -> JSONResponse:
+    def _handle_business_error(
+        self, request: Request, exc: Exception, error_name: str
+    ) -> JSONResponse:
         """Handle known business errors with standardized response"""
-        status_code, error_type, default_message = self._business_error_mapping[error_name]
+        status_code, error_type, default_message = self._business_error_mapping[
+            error_name
+        ]
 
         # Use exception message if available, otherwise use default
         detail = str(exc) if str(exc) else default_message
@@ -348,7 +396,11 @@ class BusinessErrorMiddleware(BaseHTTPMiddleware):
             request=request,
         )
 
-        return JSONResponse(status_code=status_code, content=error_response, headers=self._get_security_headers())
+        return JSONResponse(
+            status_code=status_code,
+            content=error_response,
+            headers=self._get_security_headers(),
+        )
 
     def _get_security_headers(self) -> Dict[str, str]:
         """Get security headers for business error responses"""