12
12
13
13
14
14
class ProxyEventType (Enum ):
15
+ """An enumerations of the supported proxy event types.
16
+
17
+ **NOTE:** api_gateway is an alias of http_api_v1"""
18
+
15
19
http_api_v1 = "APIGatewayProxyEvent"
16
20
http_api_v2 = "APIGatewayProxyEventV2"
17
21
alb_event = "ALBEvent"
18
22
api_gateway = http_api_v1
19
23
20
24
21
25
class CORSConfig (object ):
22
- """CORS Config"""
26
+ """CORS Config
27
+
28
+
29
+ Examples
30
+ --------
31
+
32
+ Simple cors example using the default permissive cors, not this should only be used during early prototyping
33
+
34
+ >>> from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver
35
+ >>>
36
+ >>> app = ApiGatewayResolver()
37
+ >>>
38
+ >>> @app.get("/my/path", cors=True)
39
+ >>> def with_cors():
40
+ >>> return {"message": "Foo"}
41
+
42
+ Using a custom CORSConfig where `with_cors` used the custom provided CORSConfig and `without_cors`
43
+ do not include any cors headers.
44
+
45
+ >>> from aws_lambda_powertools.event_handler.api_gateway import (
46
+ >>> ApiGatewayResolver, CORSConfig
47
+ >>> )
48
+ >>>
49
+ >>> cors_config = CORSConfig(
50
+ >>> allow_origin="https://wwww.example.com/",
51
+ >>> expose_headers=["x-exposed-response-header"],
52
+ >>> allow_headers=["x-custom-request-header"],
53
+ >>> max_age=100,
54
+ >>> allow_credentials=True,
55
+ >>> )
56
+ >>> app = ApiGatewayResolver(cors=cors_config)
57
+ >>>
58
+ >>> @app.get("/my/path", cors=True)
59
+ >>> def with_cors():
60
+ >>> return {"message": "Foo"}
61
+ >>>
62
+ >>> @app.get("/another-one")
63
+ >>> def without_cors():
64
+ >>> return {"message": "Foo"}
65
+ """
23
66
24
67
_REQUIRED_HEADERS = ["Authorization" , "Content-Type" , "X-Amz-Date" , "X-Api-Key" , "X-Amz-Security-Token" ]
25
68
@@ -55,6 +98,7 @@ def __init__(
55
98
self .allow_credentials = allow_credentials
56
99
57
100
def to_dict (self ) -> Dict [str , str ]:
101
+ """Builds the configured Access-Control http headers"""
58
102
headers = {
59
103
"Access-Control-Allow-Origin" : self .allow_origin ,
60
104
"Access-Control-Allow-Headers" : "," .join (sorted (self .allow_headers )),
@@ -68,22 +112,26 @@ def to_dict(self) -> Dict[str, str]:
68
112
return headers
69
113
70
114
71
- class Route :
72
- def __init__ (
73
- self , method : str , rule : Any , func : Callable , cors : bool , compress : bool , cache_control : Optional [str ]
74
- ):
75
- self .method = method .upper ()
76
- self .rule = rule
77
- self .func = func
78
- self .cors = cors
79
- self .compress = compress
80
- self .cache_control = cache_control
81
-
82
-
83
115
class Response :
116
+ """Response data class that provides greater control over what is returned from the proxy event"""
117
+
84
118
def __init__ (
85
119
self , status_code : int , content_type : Optional [str ], body : Union [str , bytes , None ], headers : Dict = None
86
120
):
121
+ """
122
+
123
+ Parameters
124
+ ----------
125
+ status_code: int
126
+ Http status code, example 200
127
+ content_type: str
128
+ Optionally set the Content-Type header, example "application/json". Note this will be merged into any
129
+ provided http headers
130
+ body: Union[str, bytes, None]
131
+ Optionally set the response body. Note: bytes body will be automatically base64 encoded
132
+ headers: dict
133
+ Optionally set specific http headers. Setting "Content-Type" hear would override the `content_type` value.
134
+ """
87
135
self .status_code = status_code
88
136
self .body = body
89
137
self .base64_encoded = False
@@ -92,18 +140,37 @@ def __init__(
92
140
self .headers .setdefault ("Content-Type" , content_type )
93
141
94
142
143
+ class Route :
144
+ """Internally used Route Configuration"""
145
+
146
+ def __init__ (
147
+ self , method : str , rule : Any , func : Callable , cors : bool , compress : bool , cache_control : Optional [str ]
148
+ ):
149
+ self .method = method .upper ()
150
+ self .rule = rule
151
+ self .func = func
152
+ self .cors = cors
153
+ self .compress = compress
154
+ self .cache_control = cache_control
155
+
156
+
95
157
class ResponseBuilder :
158
+ """Internally used Response builder"""
159
+
96
160
def __init__ (self , response : Response , route : Route = None ):
97
161
self .response = response
98
162
self .route = route
99
163
100
164
def _add_cors (self , cors : CORSConfig ):
165
+ """Update headers to include the configured Access-Control headers"""
101
166
self .response .headers .update (cors .to_dict ())
102
167
103
168
def _add_cache_control (self , cache_control : str ):
169
+ """Set the specified cache control headers for 200 http responses. For non-200 `no-cache` is used."""
104
170
self .response .headers ["Cache-Control" ] = cache_control if self .response .status_code == 200 else "no-cache"
105
171
106
172
def _compress (self ):
173
+ """Compress the response body, but only if `Accept-Encoding` headers includes gzip."""
107
174
self .response .headers ["Content-Encoding" ] = "gzip"
108
175
if isinstance (self .response .body , str ):
109
176
self .response .body = bytes (self .response .body , "utf-8" )
@@ -122,6 +189,7 @@ def _route(self, event: BaseProxyEvent, cors: Optional[CORSConfig]):
122
189
self ._compress ()
123
190
124
191
def build (self , event : BaseProxyEvent , cors : CORSConfig = None ) -> Dict [str , Any ]:
192
+ """Build the full response dict to be returned by the lambda"""
125
193
self ._route (event , cors )
126
194
127
195
if isinstance (self .response .body , bytes ):
0 commit comments