feat(event_handler): add extras HTTP Error Code Exceptions (#6454)

* feat(event_handler): add error status

* update doc with new errors

* fix typo

* change class description

---------

Co-authored-by: Ana Falcão <[email protected]>
Co-authored-by: Ana Falcao <[email protected]>
Co-authored-by: Leandro Damascena <[email protected]>

Author: 3 people

aws_lambda_powertools/event_handler/exceptions.py

@@ -2,7 +2,7 @@
 
 
 class ServiceError(Exception):
-    """API Gateway and ALB HTTP Service Error"""
+    """Powertools class HTTP Service Error"""
 
     def __init__(self, status_code: int, msg: str):
         """
@@ -18,28 +18,56 @@ def __init__(self, status_code: int, msg: str):
 
 
 class BadRequestError(ServiceError):
-    """API Gateway and ALB Bad Request Error (400)"""
+    """Powertools class Bad Request Error (400)"""
 
     def __init__(self, msg: str):
         super().__init__(HTTPStatus.BAD_REQUEST, msg)
 
 
 class UnauthorizedError(ServiceError):
-    """API Gateway and ALB Unauthorized Error (401)"""
+    """Powertools class Unauthorized Error (401)"""
 
     def __init__(self, msg: str):
         super().__init__(HTTPStatus.UNAUTHORIZED, msg)
 
 
+class ForbiddenError(ServiceError):
+    """Powertools class Forbidden Error (403)"""
+
+    def __init__(self, msg: str):
+        super().__init__(HTTPStatus.FORBIDDEN, msg)
+
+
 class NotFoundError(ServiceError):
-    """API Gateway and ALB Not Found Error (404)"""
+    """Powertools class Not Found Error (404)"""
 
     def __init__(self, msg: str = "Not found"):
         super().__init__(HTTPStatus.NOT_FOUND, msg)
 
 
+class RequestTimeoutError(ServiceError):
+    """Powertools class Request Timeout Error (408)"""
+
+    def __init__(self, msg: str):
+        super().__init__(HTTPStatus.REQUEST_TIMEOUT, msg)
+
+
+class RequestEntityTooLargeError(ServiceError):
+    """Powertools class Request Entity Too Large Error (413)"""
+
+    def __init__(self, msg: str):
+        super().__init__(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, msg)
+
+
 class InternalServerError(ServiceError):
-    """API Gateway and ALB Internal Server Error (500)"""
+    """Powertools class Internal Server Error (500)"""
 
     def __init__(self, message: str):
         super().__init__(HTTPStatus.INTERNAL_SERVER_ERROR, message)
+
+
+class ServiceUnavailableError(ServiceError):
+    """Powertools class Service Unavailable Error (503)"""
+
+    def __init__(self, msg: str):
+        super().__init__(HTTPStatus.SERVICE_UNAVAILABLE, msg)

docs/core/event_handler/api_gateway.md

@@ -595,9 +595,9 @@ You can easily raise any HTTP Error back to the client using `ServiceError` exce
 ???+ info
     If you need to send custom headers, use [Response](#fine-grained-responses) class instead.
 
-We provide pre-defined errors for the most popular ones such as HTTP 400, 401, 404, 500.
+We provide pre-defined errors for the most popular ones based on [AWS Lambda API Reference Common Erros](https://docs.aws.amazon.com/lambda/latest/api/CommonErrors.html).
 
-```python hl_lines="6-11 23 28 33 38 43" title="Raising common HTTP Status errors (4xx, 5xx)"
+```python hl_lines="7-15 27 32 37 42 47 52 57 62 67" title="Raising common HTTP Status errors (4xx, 5xx)"
 --8<-- "examples/event_handler_rest/src/raising_http_errors.py"
 ```
 

examples/event_handler_rest/src/raising_http_errors.py

@@ -5,9 +5,13 @@
 from aws_lambda_powertools.event_handler import APIGatewayRestResolver
 from aws_lambda_powertools.event_handler.exceptions import (
     BadRequestError,
+    ForbiddenError,
     InternalServerError,
     NotFoundError,
+    RequestEntityTooLargeError,
+    RequestTimeoutError,
     ServiceError,
+    ServiceUnavailableError,
     UnauthorizedError,
 )
 from aws_lambda_powertools.logging import correlation_paths
@@ -28,21 +32,40 @@ def unauthorized_error():
     raise UnauthorizedError("Unauthorized")  # HTTP 401
 
 
+@app.get(rule="/forbidden-error")
+def forbidden_error():
+    raise ForbiddenError("Access denied")  # HTTP 403
+
+
 @app.get(rule="/not-found-error")
 def not_found_error():
     raise NotFoundError  # HTTP 404
 
 
+@app.get(rule="/request-timeout-error")
+def request_timeout_error():
+    raise RequestTimeoutError("Request timed out")  # HTTP 408
+
+
 @app.get(rule="/internal-server-error")
 def internal_server_error():
     raise InternalServerError("Internal server error")  # HTTP 500
 
 
+@app.get(rule="/request-entity-too-large-error")
+def request_entity_too_large_error():
+    raise RequestEntityTooLargeError("Request payload too large")  # HTTP 413
+
+
 @app.get(rule="/service-error", cors=True)
 def service_error():
     raise ServiceError(502, "Something went wrong!")
 
 
+@app.get(rule="/service-unavailable-error")
+def service_unavailable_error():
+    raise ServiceUnavailableError("Service is temporarily unavailable")  # HTTP 503
+
 @app.get("/todos")
 @tracer.capture_method
 def get_todos():

tests/functional/event_handler/required_dependencies/test_api_gateway.py

@@ -26,9 +26,13 @@
 )
 from aws_lambda_powertools.event_handler.exceptions import (
     BadRequestError,
+    ForbiddenError,
     InternalServerError,
     NotFoundError,
+    RequestEntityTooLargeError,
+    RequestTimeoutError,
     ServiceError,
+    ServiceUnavailableError,
     UnauthorizedError,
 )
 from aws_lambda_powertools.shared import constants
@@ -873,6 +877,21 @@ def unauthorized_error():
     expected = {"statusCode": 401, "message": "Unauthorized"}
     assert result["body"] == json_dump(expected)
 
+    # GIVEN a ForbiddenError
+    @app.get(rule="/forbidden-error", cors=False)
+    def forbidden_error():
+        raise ForbiddenError("Access denied")
+
+    # WHEN calling the handler
+    # AND path is /forbidden-error
+    result = app({"path": "/forbidden-error", "httpMethod": "GET"}, None)
+    # THEN return the forbidden error response
+    # AND status code equals 403
+    assert result["statusCode"] == 403
+    assert result["multiValueHeaders"]["Content-Type"] == [content_types.APPLICATION_JSON]
+    expected = {"statusCode": 403, "message": "Access denied"}
+    assert result["body"] == json_dump(expected)
+
     # GIVEN an NotFoundError
     @app.get(rule="/not-found-error", cors=False)
     def not_found_error():
@@ -888,6 +907,36 @@ def not_found_error():
     expected = {"statusCode": 404, "message": "Not found"}
     assert result["body"] == json_dump(expected)
 
+    # GIVEN a RequestTimeoutError
+    @app.get(rule="/request-timeout-error", cors=False)
+    def request_timeout_error():
+        raise RequestTimeoutError("Request timed out")
+
+    # WHEN calling the handler
+    # AND path is /request-timeout-error
+    result = app({"path": "/request-timeout-error", "httpMethod": "GET"}, None)
+    # THEN return the request timeout error response
+    # AND status code equals 408
+    assert result["statusCode"] == 408
+    assert result["multiValueHeaders"]["Content-Type"] == [content_types.APPLICATION_JSON]
+    expected = {"statusCode": 408, "message": "Request timed out"}
+    assert result["body"] == json_dump(expected)
+
+    # GIVEN a RequestEntityTooLargeError
+    @app.get(rule="/request-entity-too-large-error", cors=False)
+    def request_entity_too_large_error():
+        raise RequestEntityTooLargeError("Request payload too large")
+
+    # WHEN calling the handler
+    # AND path is /request-entity-too-large-error
+    result = app({"path": "/request-entity-too-large-error", "httpMethod": "GET"}, None)
+    # THEN return the request entity too large error response
+    # AND status code equals 413
+    assert result["statusCode"] == 413
+    assert result["multiValueHeaders"]["Content-Type"] == [content_types.APPLICATION_JSON]
+    expected = {"statusCode": 413, "message": "Request payload too large"}
+    assert result["body"] == json_dump(expected)
+
     # GIVEN an InternalServerError
     @app.get(rule="/internal-server-error", cors=False)
     def internal_server_error():
@@ -903,6 +952,21 @@ def internal_server_error():
     expected = {"statusCode": 500, "message": "Internal server error"}
     assert result["body"] == json_dump(expected)
 
+    # GIVEN a ServiceUnavailableError
+    @app.get(rule="/service-unavailable-error", cors=False)
+    def service_unavailable_error():
+        raise ServiceUnavailableError("Service is temporarily unavailable")
+
+    # WHEN calling the handler
+    # AND path is /service-unavailable-error
+    result = app({"path": "/service-unavailable-error", "httpMethod": "GET"}, None)
+    # THEN return the service unavailable error response
+    # AND status code equals 503
+    assert result["statusCode"] == 503
+    assert result["multiValueHeaders"]["Content-Type"] == [content_types.APPLICATION_JSON]
+    expected = {"statusCode": 503, "message": "Service is temporarily unavailable"}
+    assert result["body"] == json_dump(expected)
+
     # GIVEN an ServiceError with a custom status code
     @app.get(rule="/service-error")
     def service_error():