AppSync

CLASS	DESCRIPTION
AppSyncResolver	AppSync GraphQL API Resolver

AppSyncResolver

AppSyncResolver()

Bases: Router

AppSync GraphQL API Resolver

Example

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())

METHOD	DESCRIPTION
batch_resolver	Registers batch resolver function for GraphQL type and field name.
exception_handler	A decorator function that registers a handler for one or more exception types.
include_router	Adds all resolvers defined in a router
resolve	Resolves the response based on the provide event and decorator routes
resolver	Registers direct resolver function for GraphQL type and field name.

Source code in aws_lambda_powertools/event_handler/appsync.py

def __init__(self):
    """
    Initialize a new instance of the AppSyncResolver.
    """
    super().__init__()
    self.context = {}  # early init as customers might add context before event resolution
    self._exception_handlers: dict[type, Callable] = {}

batch_resolver

batch_resolver(
    type_name: str = "*",
    field_name: str | None = None,
    raise_on_error: bool = False,
    aggregate: bool = True,
) -> Callable

Registers batch resolver function for GraphQL type and field name.

By default, we handle errors gracefully by returning None. If you want to short-circuit and fail the entire batch use raise_on_error=True.

PARAMETER	DESCRIPTION
type_name	GraphQL type e.g., Query, Mutation, by default "*" meaning any TYPE: str DEFAULT: '*'
field_name	GraphQL field e.g., getTodo, createTodo, by default None TYPE: Optional[str] DEFAULT: None
raise_on_error	Whether to fail entire batch upon error, or handle errors gracefully (None), by default False TYPE: bool DEFAULT: False
aggregate	A flag indicating whether the batch items should be processed at once or individually. If True (default), the batch resolver will process all items in the batch as a single event. If False, the batch resolver will process each item in the batch individually. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
Callable	Registered resolver

Source code in aws_lambda_powertools/event_handler/appsync.py

def batch_resolver(
    self,
    type_name: str = "*",
    field_name: str | None = None,
    raise_on_error: bool = False,
    aggregate: bool = True,
) -> Callable:
    """Registers batch resolver function for GraphQL type and field name.

    By default, we handle errors gracefully by returning `None`. If you want
    to short-circuit and fail the entire batch use `raise_on_error=True`.

    Parameters
    ----------
    type_name : str, optional
        GraphQL type e.g., Query, Mutation, by default "*" meaning any
    field_name : Optional[str], optional
        GraphQL field e.g., getTodo, createTodo, by default None
    raise_on_error : bool, optional
        Whether to fail entire batch upon error, or handle errors gracefully (None), by default False
    aggregate: bool
        A flag indicating whether the batch items should be processed at once or individually.
        If True (default), the batch resolver will process all items in the batch as a single event.
        If False, the batch resolver will process each item in the batch individually.

    Returns
    -------
    Callable
        Registered resolver
    """
    return self._batch_resolver_registry.register(
        field_name=field_name,
        type_name=type_name,
        raise_on_error=raise_on_error,
        aggregate=aggregate,
    )

exception_handler

exception_handler(
    exc_class: type[Exception] | list[type[Exception]],
)

A decorator function that registers a handler for one or more exception types.

PARAMETER	DESCRIPTION
exc_class	A single exception type or a list of exception types. TYPE: type[Exception] | list[type[Exception]]

RETURNS	DESCRIPTION
Callable	A decorator function that registers the exception handler.

Source code in aws_lambda_powertools/event_handler/appsync.py

def exception_handler(self, exc_class: type[Exception] | list[type[Exception]]):
    """
    A decorator function that registers a handler for one or more exception types.

    Parameters
    ----------
    exc_class (type[Exception] | list[type[Exception]])
        A single exception type or a list of exception types.

    Returns
    -------
    Callable:
        A decorator function that registers the exception handler.
    """

    def register_exception_handler(func: Callable):
        if isinstance(exc_class, list):  # pragma: no cover
            for exp in exc_class:
                self._exception_handlers[exp] = func
        else:
            self._exception_handlers[exc_class] = func
        return func

    return register_exception_handler

include_router

include_router(router: Router) -> None

Adds all resolvers defined in a router

PARAMETER	DESCRIPTION
router	A router containing a dict of field resolvers TYPE: Router

Source code in aws_lambda_powertools/event_handler/appsync.py

def include_router(self, router: Router) -> None:
    """Adds all resolvers defined in a router

    Parameters
    ----------
    router : Router
        A router containing a dict of field resolvers
    """

    # Merge app and router context
    logger.debug("Merging router and app context")
    self.context.update(**router.context)

    # use pointer to allow context clearance after event is processed e.g., resolve(evt, ctx)
    router.context = self.context

    logger.debug("Merging router resolver registries")
    self._resolver_registry.merge(router._resolver_registry)
    self._batch_resolver_registry.merge(router._batch_resolver_registry)
    self._async_batch_resolver_registry.merge(router._async_batch_resolver_registry)

resolve

resolve(
    event: dict | list[dict],
    context: LambdaContext,
    data_model: type[
        AppSyncResolverEvent
    ] = AppSyncResolverEvent,
) -> Any

Resolves the response based on the provide event and decorator routes

PARAMETER	DESCRIPTION
event	Lambda event either coming from batch processing endpoint or from standard processing endpoint TYPE: dict | list[Dict]
context	Lambda context TYPE: LambdaContext
data_model	Your data data_model to decode AppSync event, by default AppSyncResolverEvent TYPE: type[AppSyncResolverEvent] 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	DESCRIPTION
Any	Returns the result of the resolver

RAISES	DESCRIPTION
ValueError	If we could not find a field resolver

Source code in aws_lambda_powertools/event_handler/appsync.py

def resolve(
    self,
    event: dict | list[dict],
    context: LambdaContext,
    data_model: type[AppSyncResolverEvent] = AppSyncResolverEvent,
) -> Any:
    """Resolves the response based on the provide event and decorator routes

    Parameters
    ----------
    event : dict | list[Dict]
        Lambda event either coming from batch processing endpoint or from standard processing endpoint
    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.lambda_context = context
    Router.lambda_context = context

    try:
        if isinstance(event, list):
            Router.current_batch_event = [data_model(e) for e in event]
            response = self._call_batch_resolver(event=event, data_model=data_model)
        else:
            Router.current_event = data_model(event)
            response = self._call_single_resolver(event=event, data_model=data_model)
    except Exception as exp:
        response_builder = self._lookup_exception_handler(type(exp))
        if response_builder:
            return response_builder(exp)
        raise

    # We don't clear the context for coroutines because we don't have control over the event loop.
    # If we clean the context immediately, it might not be available when the coroutine is actually executed.
    # For single async operations, the context should be cleaned up manually after the coroutine completes.
    # See: https://github.com/aws-powertools/powertools-lambda-python/issues/5290
    # REVIEW: Review this support in Powertools V4
    if not asyncio.iscoroutine(response):
        self.clear_context()

    return response

resolver

resolver(
    type_name: str = "*", field_name: str | None = None
) -> Callable

Registers direct resolver function for GraphQL type and field name.

PARAMETER	DESCRIPTION
type_name	GraphQL type e.g., Query, Mutation, by default "*" meaning any TYPE: str DEFAULT: '*'
field_name	GraphQL field e.g., getTodo, createTodo, by default None TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
Callable	Registered resolver

Example

from aws_lambda_powertools.event_handler import AppSyncResolver

from typing import TypedDict

app = AppSyncResolver()

class Todo(TypedDict, total=False):
    id: str
    userId: str
    title: str
    completed: bool

# resolve any GraphQL `getTodo` queries
# arguments are injected as function arguments as-is
@app.resolver(type_name="Query", field_name="getTodo")
def get_todo(id: str = "", status: str = "open") -> Todo:
    todos: Response = requests.get(f"https://jsonplaceholder.typicode.com/todos/{id}")
    todos.raise_for_status()

    return todos.json()

def lambda_handler(event, context):
    return app.resolve(event, context)

Source code in aws_lambda_powertools/event_handler/appsync.py

def resolver(self, type_name: str = "*", field_name: str | None = None) -> Callable:
    """Registers direct resolver function for GraphQL type and field name.

    Parameters
    ----------
    type_name : str, optional
        GraphQL type e.g., Query, Mutation, by default "*" meaning any
    field_name : Optional[str], optional
        GraphQL field e.g., getTodo, createTodo, by default None

    Returns
    -------
    Callable
        Registered resolver

    Example
    -------

    ```python
    from aws_lambda_powertools.event_handler import AppSyncResolver

    from typing import TypedDict

    app = AppSyncResolver()

    class Todo(TypedDict, total=False):
        id: str
        userId: str
        title: str
        completed: bool

    # resolve any GraphQL `getTodo` queries
    # arguments are injected as function arguments as-is
    @app.resolver(type_name="Query", field_name="getTodo")
    def get_todo(id: str = "", status: str = "open") -> Todo:
        todos: Response = requests.get(f"https://jsonplaceholder.typicode.com/todos/{id}")
        todos.raise_for_status()

        return todos.json()

    def lambda_handler(event, context):
        return app.resolve(event, context)
    ```
    """
    return self._resolver_registry.register(field_name=field_name, type_name=type_name)