Module aws_lambda_powertools.event_handler
Event handler decorators for common Lambda events
Expand source code
"""
Event handler decorators for common Lambda events
"""
from .api_gateway import ApiGatewayResolver
from .appsync import AppSyncResolver
__all__ = ["AppSyncResolver", "ApiGatewayResolver"]
Sub-modules
aws_lambda_powertools.event_handler.api_gateway
aws_lambda_powertools.event_handler.appsync
aws_lambda_powertools.event_handler.content_types
aws_lambda_powertools.event_handler.exceptions
Classes
class ApiGatewayResolver (proxy_type: enum.Enum = ProxyEventType.APIGatewayProxyEvent, cors: Optional[CORSConfig] = None, debug: Optional[bool] = None, serializer: Optional[Callable[[Dict[~KT, ~VT]], str]] = None)
-
API Gateway and ALB proxy resolver
Examples
Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator
from aws_lambda_powertools import Tracer from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver tracer = Tracer() app = ApiGatewayResolver() @app.get("/get-call") def simple_get(): return {"message": "Foo"} @app.post("/post-call") def simple_post(): post_data: dict = app.current_event.json_body return {"message": post_data["value"]} @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context)
Parameters
proxy_type
:ProxyEventType
- Proxy request type, defaults to API Gateway V1
cors
:CORSConfig
- Optionally configure and enabled CORS. Not each route will need to have to cors=True
debug
:Optional[bool]
- Enables debug mode, by default False. Can be also be enabled by "POWERTOOLS_EVENT_HANDLER_DEBUG" environment variable
Expand source code
class ApiGatewayResolver: """API Gateway and ALB proxy resolver Examples -------- Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator ```python from aws_lambda_powertools import Tracer from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver tracer = Tracer() app = ApiGatewayResolver() @app.get("/get-call") def simple_get(): return {"message": "Foo"} @app.post("/post-call") def simple_post(): post_data: dict = app.current_event.json_body return {"message": post_data["value"]} @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context) ``` """ current_event: BaseProxyEvent lambda_context: LambdaContext def __init__( self, proxy_type: Enum = ProxyEventType.APIGatewayProxyEvent, cors: Optional[CORSConfig] = None, debug: Optional[bool] = None, serializer: Optional[Callable[[Dict], str]] = None, ): """ Parameters ---------- proxy_type: ProxyEventType Proxy request type, defaults to API Gateway V1 cors: CORSConfig Optionally configure and enabled CORS. Not each route will need to have to cors=True debug: Optional[bool] Enables debug mode, by default False. Can be also be enabled by "POWERTOOLS_EVENT_HANDLER_DEBUG" environment variable """ self._proxy_type = proxy_type self._routes: List[Route] = [] self._cors = cors self._cors_enabled: bool = cors is not None self._cors_methods: Set[str] = {"OPTIONS"} self._debug = resolve_truthy_env_var_choice( env=os.getenv(constants.EVENT_HANDLER_DEBUG_ENV, "false"), choice=debug ) # Allow for a custom serializer or a concise json serialization self._serializer = serializer or partial(json.dumps, separators=(",", ":"), cls=Encoder) if self._debug: # Always does a pretty print when in debug mode self._serializer = partial(json.dumps, indent=4, cls=Encoder) def get(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None): """Get route decorator with GET `method` Examples -------- Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator ```python from aws_lambda_powertools import Tracer from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver tracer = Tracer() app = ApiGatewayResolver() @app.get("/get-call") def simple_get(): return {"message": "Foo"} @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context) ``` """ return self.route(rule, "GET", cors, compress, cache_control) def post(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None): """Post route decorator with POST `method` Examples -------- Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator ```python from aws_lambda_powertools import Tracer from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver tracer = Tracer() app = ApiGatewayResolver() @app.post("/post-call") def simple_post(): post_data: dict = app.current_event.json_body return {"message": post_data["value"]} @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context) ``` """ return self.route(rule, "POST", cors, compress, cache_control) def put(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None): """Put route decorator with PUT `method` Examples -------- Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator ```python from aws_lambda_powertools import Tracer from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver tracer = Tracer() app = ApiGatewayResolver() @app.put("/put-call") def simple_put(): put_data: dict = app.current_event.json_body return {"message": put_data["value"]} @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context) ``` """ return self.route(rule, "PUT", cors, compress, cache_control) def delete( self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None ): """Delete route decorator with DELETE `method` Examples -------- Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator ```python from aws_lambda_powertools import Tracer from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver tracer = Tracer() app = ApiGatewayResolver() @app.delete("/delete-call") def simple_delete(): return {"message": "deleted"} @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context) ``` """ return self.route(rule, "DELETE", cors, compress, cache_control) def patch( self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None ): """Patch route decorator with PATCH `method` Examples -------- Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator ```python from aws_lambda_powertools import Tracer from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver tracer = Tracer() app = ApiGatewayResolver() @app.patch("/patch-call") def simple_patch(): patch_data: dict = app.current_event.json_body patch_data["value"] = patched return {"message": patch_data} @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context) ``` """ return self.route(rule, "PATCH", cors, compress, cache_control) def route( self, rule: str, method: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None, ): """Route decorator includes parameter `method`""" def register_resolver(func: Callable): logger.debug(f"Adding route using rule {rule} and method {method.upper()}") if cors is None: cors_enabled = self._cors_enabled else: cors_enabled = cors self._routes.append(Route(method, self._compile_regex(rule), func, cors_enabled, compress, cache_control)) if cors_enabled: logger.debug(f"Registering method {method.upper()} to Allow Methods in CORS") self._cors_methods.add(method.upper()) return func return register_resolver def resolve(self, event, context) -> Dict[str, Any]: """Resolves the response based on the provide event and decorator routes Parameters ---------- event: Dict[str, Any] Event context: LambdaContext Lambda context Returns ------- dict Returns the dict response """ if self._debug: print(self._json_dump(event)) self.current_event = self._to_proxy_event(event) self.lambda_context = context return self._resolve().build(self.current_event, self._cors) def __call__(self, event, context) -> Any: return self.resolve(event, context) @staticmethod def _compile_regex(rule: str): """Precompile regex pattern Logic ----- 1. Find any dynamic routes defined as <pattern> e.g. @app.get("/accounts/<account_id>") 2. Create a new regex by substituting every dynamic route found as a named group (?P<group>), and match whole words only (word boundary) instead of a greedy match non-greedy example with word boundary rule: '/accounts/<account_id>' regex: r'/accounts/(?P<account_id>\\w+\\b)' value: /accounts/123/some_other_path account_id: 123 greedy example without word boundary regex: r'/accounts/(?P<account_id>.+)' value: /accounts/123/some_other_path account_id: 123/some_other_path 3. Compiles a regex and include start (^) and end ($) in between for an exact match NOTE: See #520 for context """ rule_regex: str = re.sub(_DYNAMIC_ROUTE_PATTERN, _NAMED_GROUP_BOUNDARY_PATTERN, rule) return re.compile("^{}$".format(rule_regex)) def _to_proxy_event(self, event: Dict) -> BaseProxyEvent: """Convert the event dict to the corresponding data class""" if self._proxy_type == ProxyEventType.APIGatewayProxyEvent: logger.debug("Converting event to API Gateway REST API contract") return APIGatewayProxyEvent(event) if self._proxy_type == ProxyEventType.APIGatewayProxyEventV2: logger.debug("Converting event to API Gateway HTTP API contract") return APIGatewayProxyEventV2(event) logger.debug("Converting event to ALB contract") return ALBEvent(event) def _resolve(self) -> ResponseBuilder: """Resolves the response or return the not found response""" method = self.current_event.http_method.upper() path = self.current_event.path for route in self._routes: if method != route.method: continue match: Optional[re.Match] = route.rule.match(path) if match: logger.debug("Found a registered route. Calling function") return self._call_route(route, match.groupdict()) # pass fn args logger.debug(f"No match found for path {path} and method {method}") return self._not_found(method) def _not_found(self, method: str) -> ResponseBuilder: """Called when no matching route was found and includes support for the cors preflight response""" headers = {} if self._cors: logger.debug("CORS is enabled, updating headers.") headers.update(self._cors.to_dict()) if method == "OPTIONS": logger.debug("Pre-flight request detected. Returning CORS with null response") headers["Access-Control-Allow-Methods"] = ",".join(sorted(self._cors_methods)) return ResponseBuilder(Response(status_code=204, content_type=None, headers=headers, body=None)) return ResponseBuilder( Response( status_code=HTTPStatus.NOT_FOUND.value, content_type=content_types.APPLICATION_JSON, headers=headers, body=self._json_dump({"statusCode": HTTPStatus.NOT_FOUND.value, "message": "Not found"}), ) ) def _call_route(self, route: Route, args: Dict[str, str]) -> ResponseBuilder: """Actually call the matching route with any provided keyword arguments.""" try: return ResponseBuilder(self._to_response(route.func(**args)), route) except ServiceError as e: return ResponseBuilder( Response( status_code=e.status_code, content_type=content_types.APPLICATION_JSON, body=self._json_dump({"statusCode": e.status_code, "message": e.msg}), ), route, ) except Exception: if self._debug: # If the user has turned on debug mode, # we'll let the original exception propagate so # they get more information about what went wrong. return ResponseBuilder( Response( status_code=500, content_type=content_types.TEXT_PLAIN, body="".join(traceback.format_exc()), ) ) raise def _to_response(self, result: Union[Dict, Response]) -> Response: """Convert the route's result to a Response 2 main result types are supported: - Dict[str, Any]: Rest api response with just the Dict to json stringify and content-type is set to application/json - Response: returned as is, and allows for more flexibility """ if isinstance(result, Response): return result logger.debug("Simple response detected, serializing return before constructing final response") return Response( status_code=200, content_type=content_types.APPLICATION_JSON, body=self._json_dump(result), ) def _json_dump(self, obj: Any) -> str: return self._serializer(obj)
Class variables
var current_event : BaseProxyEvent
var lambda_context : LambdaContext
Methods
def delete(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None)
-
Delete route decorator with DELETE
method
Examples
Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator
from aws_lambda_powertools import Tracer from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver tracer = Tracer() app = ApiGatewayResolver() @app.delete("/delete-call") def simple_delete(): return {"message": "deleted"} @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context)
Expand source code
def delete( self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None ): """Delete route decorator with DELETE `method` Examples -------- Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator ```python from aws_lambda_powertools import Tracer from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver tracer = Tracer() app = ApiGatewayResolver() @app.delete("/delete-call") def simple_delete(): return {"message": "deleted"} @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context) ``` """ return self.route(rule, "DELETE", cors, compress, cache_control)
def get(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None)
-
Get route decorator with GET
method
Examples
Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator
from aws_lambda_powertools import Tracer from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver tracer = Tracer() app = ApiGatewayResolver() @app.get("/get-call") def simple_get(): return {"message": "Foo"} @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context)
Expand source code
def get(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None): """Get route decorator with GET `method` Examples -------- Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator ```python from aws_lambda_powertools import Tracer from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver tracer = Tracer() app = ApiGatewayResolver() @app.get("/get-call") def simple_get(): return {"message": "Foo"} @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context) ``` """ return self.route(rule, "GET", cors, compress, cache_control)
def patch(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None)
-
Patch route decorator with PATCH
method
Examples
Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator
from aws_lambda_powertools import Tracer from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver tracer = Tracer() app = ApiGatewayResolver() @app.patch("/patch-call") def simple_patch(): patch_data: dict = app.current_event.json_body patch_data["value"] = patched return {"message": patch_data} @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context)
Expand source code
def patch( self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None ): """Patch route decorator with PATCH `method` Examples -------- Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator ```python from aws_lambda_powertools import Tracer from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver tracer = Tracer() app = ApiGatewayResolver() @app.patch("/patch-call") def simple_patch(): patch_data: dict = app.current_event.json_body patch_data["value"] = patched return {"message": patch_data} @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context) ``` """ return self.route(rule, "PATCH", cors, compress, cache_control)
def post(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None)
-
Post route decorator with POST
method
Examples
Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator
from aws_lambda_powertools import Tracer from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver tracer = Tracer() app = ApiGatewayResolver() @app.post("/post-call") def simple_post(): post_data: dict = app.current_event.json_body return {"message": post_data["value"]} @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context)
Expand source code
def post(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None): """Post route decorator with POST `method` Examples -------- Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator ```python from aws_lambda_powertools import Tracer from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver tracer = Tracer() app = ApiGatewayResolver() @app.post("/post-call") def simple_post(): post_data: dict = app.current_event.json_body return {"message": post_data["value"]} @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context) ``` """ return self.route(rule, "POST", cors, compress, cache_control)
def put(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None)
-
Put route decorator with PUT
method
Examples
Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator
from aws_lambda_powertools import Tracer from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver tracer = Tracer() app = ApiGatewayResolver() @app.put("/put-call") def simple_put(): put_data: dict = app.current_event.json_body return {"message": put_data["value"]} @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context)
Expand source code
def put(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None): """Put route decorator with PUT `method` Examples -------- Simple example with a custom lambda handler using the Tracer capture_lambda_handler decorator ```python from aws_lambda_powertools import Tracer from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver tracer = Tracer() app = ApiGatewayResolver() @app.put("/put-call") def simple_put(): put_data: dict = app.current_event.json_body return {"message": put_data["value"]} @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context) ``` """ return self.route(rule, "PUT", cors, compress, cache_control)
def resolve(self, event, context) ‑> Dict[str, Any]
-
Resolves the response based on the provide event and decorator routes
Parameters
event
:Dict[str, Any]
- Event
context
:LambdaContext
- Lambda context
Returns
dict
- Returns the dict response
Expand source code
def resolve(self, event, context) -> Dict[str, Any]: """Resolves the response based on the provide event and decorator routes Parameters ---------- event: Dict[str, Any] Event context: LambdaContext Lambda context Returns ------- dict Returns the dict response """ if self._debug: print(self._json_dump(event)) self.current_event = self._to_proxy_event(event) self.lambda_context = context return self._resolve().build(self.current_event, self._cors)
def route(self, rule: str, method: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None)
-
Route decorator includes parameter
method
Expand source code
def route( self, rule: str, method: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None, ): """Route decorator includes parameter `method`""" def register_resolver(func: Callable): logger.debug(f"Adding route using rule {rule} and method {method.upper()}") if cors is None: cors_enabled = self._cors_enabled else: cors_enabled = cors self._routes.append(Route(method, self._compile_regex(rule), func, cors_enabled, compress, cache_control)) if cors_enabled: logger.debug(f"Registering method {method.upper()} to Allow Methods in CORS") self._cors_methods.add(method.upper()) return func return register_resolver
class AppSyncResolver
-
AppSync resolver decorator
Example
Sample usage
from aws_lambda_powertools.event_handler import AppSyncResolver app = AppSyncResolver() @app.resolver(type_name="Query", field_name="listLocations") def list_locations(page: int = 0, size: int = 10) -> list: # Your logic to fetch locations with arguments passed in return [{"id": 100, "name": "Smooth Grooves"}] @app.resolver(type_name="Merchant", field_name="extraInfo") def get_extra_info() -> dict: # Can use "app.current_event.source" to filter within the parent context account_type = app.current_event.source["accountType"] method = "BTC" if account_type == "NEW" else "USD" return {"preferredPaymentMethod": method} @app.resolver(field_name="commonField") def common_field() -> str: # Would match all fieldNames matching 'commonField' return str(uuid.uuid4())
Expand source code
class AppSyncResolver: """ AppSync resolver decorator Example ------- **Sample usage** from aws_lambda_powertools.event_handler import AppSyncResolver app = AppSyncResolver() @app.resolver(type_name="Query", field_name="listLocations") def list_locations(page: int = 0, size: int = 10) -> list: # Your logic to fetch locations with arguments passed in return [{"id": 100, "name": "Smooth Grooves"}] @app.resolver(type_name="Merchant", field_name="extraInfo") def get_extra_info() -> dict: # Can use "app.current_event.source" to filter within the parent context account_type = app.current_event.source["accountType"] method = "BTC" if account_type == "NEW" else "USD" return {"preferredPaymentMethod": method} @app.resolver(field_name="commonField") def common_field() -> str: # Would match all fieldNames matching 'commonField' return str(uuid.uuid4()) """ current_event: AppSyncResolverEventT # type: ignore[valid-type] lambda_context: LambdaContext def __init__(self): self._resolvers: dict = {} def resolver(self, type_name: str = "*", field_name: Optional[str] = None): """Registers the resolver for field_name Parameters ---------- type_name : str Type name field_name : str Field name """ def register_resolver(func): logger.debug(f"Adding resolver `{func.__name__}` for field `{type_name}.{field_name}`") self._resolvers[f"{type_name}.{field_name}"] = {"func": func} return func return register_resolver def resolve( self, event: dict, context: LambdaContext, data_model: Type[AppSyncResolverEvent] = AppSyncResolverEvent ) -> Any: """Resolve field_name Parameters ---------- event : dict Lambda event context : LambdaContext Lambda context data_model: Your data data_model to decode AppSync event, by default AppSyncResolverEvent Example ------- ```python from aws_lambda_powertools.event_handler import AppSyncResolver from aws_lambda_powertools.utilities.typing import LambdaContext @app.resolver(field_name="createSomething") def create_something(id: str): # noqa AA03 VNE003 return id def handler(event, context: LambdaContext): return app.resolve(event, context) ``` **Bringing custom models** ```python from aws_lambda_powertools import Logger, Tracer from aws_lambda_powertools.logging import correlation_paths from aws_lambda_powertools.event_handler import AppSyncResolver tracer = Tracer(service="sample_resolver") logger = Logger(service="sample_resolver") app = AppSyncResolver() class MyCustomModel(AppSyncResolverEvent): @property def country_viewer(self) -> str: return self.request_headers.get("cloudfront-viewer-country") @app.resolver(field_name="listLocations") @app.resolver(field_name="locations") def get_locations(name: str, description: str = ""): if app.current_event.country_viewer == "US": ... return name + description @logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_RESOLVER) @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context, data_model=MyCustomModel) ``` Returns ------- Any Returns the result of the resolver Raises ------- ValueError If we could not find a field resolver """ self.current_event = data_model(event) self.lambda_context = context resolver = self._get_resolver(self.current_event.type_name, self.current_event.field_name) return resolver(**self.current_event.arguments) def _get_resolver(self, type_name: str, field_name: str) -> Callable: """Get resolver for field_name Parameters ---------- type_name : str Type name field_name : str Field name Returns ------- Callable callable function and configuration """ full_name = f"{type_name}.{field_name}" resolver = self._resolvers.get(full_name, self._resolvers.get(f"*.{field_name}")) if not resolver: raise ValueError(f"No resolver found for '{full_name}'") return resolver["func"] def __call__( self, event: dict, context: LambdaContext, data_model: Type[AppSyncResolverEvent] = AppSyncResolverEvent ) -> Any: """Implicit lambda handler which internally calls `resolve`""" return self.resolve(event, context, data_model)
Class variables
var current_event : ~AppSyncResolverEventT
var lambda_context : LambdaContext
Methods
def resolve(self, event: dict, context: LambdaContext, data_model: Type[AppSyncResolverEvent] = aws_lambda_powertools.utilities.data_classes.appsync_resolver_event.AppSyncResolverEvent) ‑> Any
-
Resolve field_name
Parameters
event
:dict
- Lambda event
context
:LambdaContext
- Lambda context
data_model: Your data data_model to decode AppSync event, by default AppSyncResolverEvent
Example
from aws_lambda_powertools.event_handler import AppSyncResolver from aws_lambda_powertools.utilities.typing import LambdaContext @app.resolver(field_name="createSomething") def create_something(id: str): # noqa AA03 VNE003 return id def handler(event, context: LambdaContext): return app.resolve(event, context)
Bringing custom models
from aws_lambda_powertools import Logger, Tracer from aws_lambda_powertools.logging import correlation_paths from aws_lambda_powertools.event_handler import AppSyncResolver tracer = Tracer(service="sample_resolver") logger = Logger(service="sample_resolver") app = AppSyncResolver() class MyCustomModel(AppSyncResolverEvent): @property def country_viewer(self) -> str: return self.request_headers.get("cloudfront-viewer-country") @app.resolver(field_name="listLocations") @app.resolver(field_name="locations") def get_locations(name: str, description: str = ""): if app.current_event.country_viewer == "US": ... return name + description @logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_RESOLVER) @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context, data_model=MyCustomModel)
Returns
Any
- Returns the result of the resolver
Raises
ValueError
- If we could not find a field resolver
Expand source code
def resolve( self, event: dict, context: LambdaContext, data_model: Type[AppSyncResolverEvent] = AppSyncResolverEvent ) -> Any: """Resolve field_name Parameters ---------- event : dict Lambda event context : LambdaContext Lambda context data_model: Your data data_model to decode AppSync event, by default AppSyncResolverEvent Example ------- ```python from aws_lambda_powertools.event_handler import AppSyncResolver from aws_lambda_powertools.utilities.typing import LambdaContext @app.resolver(field_name="createSomething") def create_something(id: str): # noqa AA03 VNE003 return id def handler(event, context: LambdaContext): return app.resolve(event, context) ``` **Bringing custom models** ```python from aws_lambda_powertools import Logger, Tracer from aws_lambda_powertools.logging import correlation_paths from aws_lambda_powertools.event_handler import AppSyncResolver tracer = Tracer(service="sample_resolver") logger = Logger(service="sample_resolver") app = AppSyncResolver() class MyCustomModel(AppSyncResolverEvent): @property def country_viewer(self) -> str: return self.request_headers.get("cloudfront-viewer-country") @app.resolver(field_name="listLocations") @app.resolver(field_name="locations") def get_locations(name: str, description: str = ""): if app.current_event.country_viewer == "US": ... return name + description @logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_RESOLVER) @tracer.capture_lambda_handler def lambda_handler(event, context): return app.resolve(event, context, data_model=MyCustomModel) ``` Returns ------- Any Returns the result of the resolver Raises ------- ValueError If we could not find a field resolver """ self.current_event = data_model(event) self.lambda_context = context resolver = self._get_resolver(self.current_event.type_name, self.current_event.field_name) return resolver(**self.current_event.arguments)
def resolver(self, type_name: str = '*', field_name: Optional[str] = None)
-
Registers the resolver for field_name
Parameters
type_name
:str
- Type name
field_name
:str
- Field name
Expand source code
def resolver(self, type_name: str = "*", field_name: Optional[str] = None): """Registers the resolver for field_name Parameters ---------- type_name : str Type name field_name : str Field name """ def register_resolver(func): logger.debug(f"Adding resolver `{func.__name__}` for field `{type_name}.{field_name}`") self._resolvers[f"{type_name}.{field_name}"] = {"func": func} return func return register_resolver