Skip to content

RFC: Create an official utility event_handler for lightweight event handlers #356

Closed
@michaelbrewer

Description

@michaelbrewer

Key information

  • RFC PR: (leave this empty)
  • Related issue(s), if known:
  • Area: Utilities
  • Meet tenets: Yes

Summary

Create an official utility event_handler for lightweight event handlers

Motivation

Unofficially we have a very lightweight resolver for AppSync measuring only 50 lines of code, but still very useful. It would be nice to have a good location for these kind of utility and they should just depend on data_classes utility

Proposal

Create a new top level utility for event_handler and then move the AppSync resolver and start to implement other useful ones like API Gateway Proxy Event Handler while ensuring we keeping them lightweight and only depend on existing Powertools libraries

AppSync in python

from typing import Any, Dict
from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.data_classes import AppSyncResolverEvent
from aws_lambda_powertools.utilities.data_classes.appsync.scalar_types_utils import (
    make_id,
    aws_date,
    aws_datetime,
    aws_time,
    aws_timestamp,
)
from aws_lambda_powertools.utilities.event_handler.appsync import (
    AppSyncResolver
)
from aws_lambda_powertools.utilities.typing import LambdaContext

logger = Logger()
app = AppSyncResolver()


@app.resolver(field_name="listLocations", include_event=True)
@app.resolver(type_name="Merchant", field_name="locations", include_event=True)
def get_locations(event: AppSyncResolverEvent, name: str = None) -> str:
    logger.info(event)
    return f"returning locations for {name}"


@app.resolver(type_name="Merchant", field_name="anotherField")
def another_field_resolver(count: int) -> Dict[str, Any]:
    return {
        "id": make_id(),
        "date": aws_date(),
        "date_time": aws_datetime(),
        "time": aws_time(),
        "ts": aws_timestamp(),
        "message": f"another_field_resolver with parameter count={count}",
    }


@app.resolver(type_name="Query", field_name="noParams")
def no_params() -> str:
    return "no_params has no params"


def handler(event: dict, context: LambdaContext) -> Any:
    return app.resolve(event, context)

API Gateway Proxy Example

from typing import Any, Dict, Tuple
from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEvent
from aws_lambda_powertools.utilities.event_handler.appsync import (
    APIGatewayProxyResolver
)
from aws_lambda_powertools.utilities.typing import LambdaContext

logger = Logger()
tracer = Tracer()
app = APIGatewayProxyResolver()


@.get("/locations/<id>", include_event=True)
def get_locations(event: APIGatewayProxyEvent, id: str) -> Tuple[str, str, str]:
    logger.info(event)
    return ("OK", "text/plain", f"returning locations by {id}")


@app.get("/locations/with-query-string")
def another_field_resolver(count: int) -> Tuple[str, str, Dict[str, Any]]:
    return (
        "OK", 
        "application/json",
        {
            "message": f"another_field_resolver with query string parameter /locations/with-query-string?count={count}",
        }
   )


@app.get("/noParams")
def no_params() -> Tuple[str, str, str]:
    return ("OK", "text/plain", "no_params has no params")


@app.post("/test/tests/<id>",  cors=True)
def print_id(id: str, body: Dict[str, Any]) -> Tuple[str, str, str]:
    return ("OK", "plain/text", id)


@logger.inject_lambda_context(correlation_id_path=correlation_paths.API_GATEWAY_REST)
@tracer.capture_lambda_handler
def handler(event: dict, context: LambdaContext) -> Dict[str, Any]:
    return app.resolve(event, context)

Drawbacks

Over time Powertools could grow quite a lot as we add more features like this, so we should start to make this more modular (or aim for this as part of V2)

Rationale and alternatives

  • What other designs have been considered? Why not them?
  • What is the impact of not doing this?

Unresolved questions

Optional, stash area for topics that need further development e.g. TBD

Metadata

Metadata

Assignees

No one assigned

    Labels

    Type

    No type

    Projects

    Status

    Triage

    Milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions