1
1
import logging
2
- from typing import Any , Callable , Dict , Optional
2
+ from typing import Dict , Optional
3
3
4
- from aws_lambda_powertools .event_handler .api_gateway import ApiGatewayResolver , Response
4
+ from aws_lambda_powertools .event_handler .api_gateway import Response
5
5
from aws_lambda_powertools .event_handler .exceptions import BadRequestError , InternalServerError
6
- from aws_lambda_powertools .event_handler .middlewares import BaseMiddlewareHandler
6
+ from aws_lambda_powertools .event_handler .middlewares import BaseMiddlewareHandler , NextMiddleware
7
+ from aws_lambda_powertools .event_handler .types import EventHandlerInstance
7
8
from aws_lambda_powertools .utilities .validation import validate
8
9
from aws_lambda_powertools .utilities .validation .exceptions import InvalidSchemaFormatError , SchemaValidationError
9
10
10
11
logger = logging .getLogger (__name__ )
11
12
12
13
13
14
class SchemaValidationMiddleware (BaseMiddlewareHandler ):
14
- """
15
- Validates the API request body against the provided schema.
15
+ """Middleware to validate API request and response against JSON Schema using the [Validation utility](https://docs.powertools.aws.dev/lambda/python/latest/utilities/validation/).
16
+
17
+ Examples
18
+ --------
19
+ **Validating incoming event**
20
+
21
+ ```python
22
+ import requests
23
+
24
+ from aws_lambda_powertools import Logger
25
+ from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Response
26
+ from aws_lambda_powertools.event_handler.middlewares import BaseMiddlewareHandler, NextMiddleware
27
+ from aws_lambda_powertools.event_handler.middlewares.schema_validation import SchemaValidationMiddleware
28
+
29
+ app = APIGatewayRestResolver()
30
+ logger = Logger()
31
+ json_schema_validation = SchemaValidationMiddleware(inbound_schema=INCOMING_JSON_SCHEMA)
32
+
16
33
34
+ @app.get("/todos", middlewares=[json_schema_validation])
35
+ def get_todos():
36
+ todos: requests.Response = requests.get("https://jsonplaceholder.typicode.com/todos")
37
+ todos.raise_for_status()
38
+
39
+ # for brevity, we'll limit to the first 10 only
40
+ return {"todos": todos.json()[:10]}
41
+
42
+
43
+ @logger.inject_lambda_context
44
+ def lambda_handler(event, context):
45
+ return app.resolve(event, context)
46
+ ```
17
47
"""
18
48
19
49
def __init__ (
@@ -23,6 +53,18 @@ def __init__(
23
53
outbound_schema : Optional [Dict ] = None ,
24
54
outbound_formats : Optional [Dict ] = None ,
25
55
):
56
+ """See [Validation utility](https://docs.powertools.aws.dev/lambda/python/latest/utilities/validation/) docs for examples on all parameters.
57
+
58
+ Parameters
59
+ ----------
60
+ inbound_schema : Dict
61
+ JSON Schema to validate incoming event
62
+ inbound_formats : Optional[Dict], optional
63
+ Custom formats containing a key (e.g. int64) and a value expressed as regex or callback returning bool, by default None
64
+ JSON Schema to validate outbound event, by default None
65
+ outbound_formats : Optional[Dict], optional
66
+ Custom formats containing a key (e.g. int64) and a value expressed as regex or callback returning bool, by default None
67
+ """ # noqa: E501
26
68
super ().__init__ ()
27
69
self .inbound_schema = inbound_schema
28
70
self .inbound_formats = inbound_formats
@@ -43,19 +85,24 @@ def bad_config(self, error: InvalidSchemaFormatError) -> Response:
43
85
logger .debug (f"Invalid Schema Format: { error } " )
44
86
raise InternalServerError ("Internal Server Error" )
45
87
46
- def handler (self , app : ApiGatewayResolver , next_middleware : Callable [..., Any ]) -> Response :
47
- """
48
- Validate using Powertools validate() utility.
49
-
50
- Return HTTP 400 Response if validation fails.
51
- Return HTTP 500 Response if validation fails due to invalid schema format.
88
+ def handler (self , app : EventHandlerInstance , next_middleware : NextMiddleware ) -> Response :
89
+ """Validates incoming JSON payload (body) against JSON Schema provided.
52
90
53
- Return the next middleware response if validation passes.
91
+ Parameters
92
+ ----------
93
+ app : EventHandlerInstance
94
+ An instance of an Event Handler
95
+ next_middleware : NextMiddleware
96
+ Callable to get response from the next middleware or route handler in the chain
54
97
55
- :param app: The ApiGatewayResolver instance
56
- :param next_middleware: The original response
57
- :return: The original response or HTTP 400 Response or HTTP 500 Response.
98
+ Returns
99
+ -------
100
+ Response
101
+ It can return three types of response objects
58
102
103
+ - Original response: Propagates HTTP response returned from the next middleware if validation succeeds
104
+ - HTTP 400: Payload or response failed JSON Schema validation
105
+ - HTTP 500: JSON Schema provided has incorrect format
59
106
"""
60
107
try :
61
108
validate (event = app .current_event .json_body , schema = self .inbound_schema , formats = self .inbound_formats )
@@ -64,8 +111,7 @@ def handler(self, app: ApiGatewayResolver, next_middleware: Callable[..., Any])
64
111
except InvalidSchemaFormatError as error :
65
112
return self .bad_config (error )
66
113
67
- # return next middleware response if validation passes.
68
- result : Response = next_middleware (app )
114
+ result = next_middleware (app )
69
115
70
116
if self .outbound_formats is not None :
71
117
try :
0 commit comments