Module aws_lambda_powertools.event_handler

Event handler decorators for common Lambda events

"""
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