Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 7 commits into from
Apr 15, 2025
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 28 additions & 0 deletions aws_lambda_powertools/event_handler/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,15 +31,43 @@ def __init__(self, msg: str):
super().__init__(HTTPStatus.UNAUTHORIZED, msg)


class ForbiddenError(ServiceError):
"""API Gateway and ALB Forbidden Error (403)"""

def __init__(self, msg: str):
super().__init__(HTTPStatus.FORBIDDEN, msg)


class NotFoundError(ServiceError):
"""API Gateway and ALB Not Found Error (404)"""

def __init__(self, msg: str = "Not found"):
super().__init__(HTTPStatus.NOT_FOUND, msg)


class RequestTimeoutError(ServiceError):
"""API Gateway and ALB Request Timeout Error (408)"""

def __init__(self, msg: str):
super().__init__(HTTPStatus.REQUEST_TIMEOUT, msg)


class RequestEntityTooLargeError(ServiceError):
"""API Gateway and ALB 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)"""

def __init__(self, message: str):
super().__init__(HTTPStatus.INTERNAL_SERVER_ERROR, message)


class ServiceUnavailableError(ServiceError):
"""API Gateway and ALB Service Unavailable Error (503)"""

def __init__(self, msg: str):
super().__init__(HTTPStatus.SERVICE_UNAVAILABLE, msg)
4 changes: 2 additions & 2 deletions docs/core/event_handler/api_gateway.md
Original file line number Diff line number Diff line change
Expand Up @@ -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"
```

Expand Down
23 changes: 23 additions & 0 deletions examples/event_handler_rest/src/raising_http_errors.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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():
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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():
Expand All @@ -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():
Expand All @@ -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():
Expand Down
Loading