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, strip_prefixes: Optional[List[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
serializer
:Callable
, optional- function to serialize
obj
to a JSON formattedstr
, by default json.dumps strip_prefixes
:List[str]
, optional- optional list of prefixes to be removed from the request path before doing the routing. This is often used with api gateways with multiple custom mappings.
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, strip_prefixes: Optional[List[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 serializer : Callable, optional function to serialize `obj` to a JSON formatted `str`, by default json.dumps strip_prefixes: List[str], optional optional list of prefixes to be removed from the request path before doing the routing. This is often used with api gateways with multiple custom mappings. """ 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 ) self._strip_prefixes = strip_prefixes # 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._remove_prefix(self.current_event.path) for route in self._routes: if method != route.method: continue match_results: Optional[re.Match] = route.rule.match(path) if match_results: logger.debug("Found a registered route. Calling function") return self._call_route(route, match_results.groupdict()) # pass fn args logger.debug(f"No match found for path {path} and method {method}") return self._not_found(method) def _remove_prefix(self, path: str) -> str: """Remove the configured prefix from the path""" if not isinstance(self._strip_prefixes, list): return path for prefix in self._strip_prefixes: if self._path_starts_with(path, prefix): return path[len(prefix) :] return path @staticmethod def _path_starts_with(path: str, prefix: str): """Returns true if the `path` starts with a prefix plus a `/`""" if not isinstance(prefix, str) or len(prefix) == 0: return False return path.startswith(prefix + "/") 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