Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(data-classes): AppSync Resolver Event (#323)
* feat(data-classes): AppSync Resolver Event * feat(data-classes): export AppSyncResolverEvent * chore: Correct the import * chore: Fix name * feat(data-classes): Add get_header_value function * feat(data-classes): Add AppSyncIdentityCognito * tests(data-classes): Add test_get_identity_object_iam * feat(logging): Add correlation path for APP_SYNC_RESOLVER * chore: Code review changes * feat(data-classes): Add AppSyncResolverEventInfo * fix(logging): Correct paths for AppSync * tests(data-classes): Add test_appsync_resolver_direct * docs(data-classes): Add AppSync Resolver docs * chore: bump ci * feat(data-classes): Add AppSyncResolverEvent.stash * refactor(data-classes): Support direct and amplify * docs(data-classes): Correct docs * docs(data-classes): Clean up docs for review * feat(data-classes): Add AppSync resolver utilities Changes: * Add helper functions to generate GraphQL scalar types * AppSyncResolver decorator which works with AppSyncResolverEvent * feat(data-classes): Include include_event and include_context * tests(data-clasess): Verify async and yield works * test(data-classes): only run async test on new python versions * test(data-classes): Verify we can support multiple mappings * chore: Update docs/utilities/data_classes.md Co-authored-by: Heitor Lessa <heitor.lessa@hotmail.com> * chore: Update docs/utilities/data_classes.md Co-authored-by: Heitor Lessa <heitor.lessa@hotmail.com> * chore: Update aws_lambda_powertools/utilities/data_classes/appsync_resolver_event.py Co-authored-by: Heitor Lessa <heitor.lessa@hotmail.com> * chore: Correct docs * chore: Correct docs * refactor(data-classes): AppSync location * docs(data-classes): Added sample usage * chore: fix docs rendering * refactor: Remove docstrings and relocate data class * docs(data-classes): Expanded on the scope and named app.py consistently Co-authored-by: Heitor Lessa <heitor.lessa@hotmail.com>
- Loading branch information
1 parent
6a9a554
commit 34c2fa9
Showing
12 changed files
with
955 additions
and
43 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Empty file.
76 changes: 76 additions & 0 deletions
76
aws_lambda_powertools/utilities/data_classes/appsync/resolver_utils.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,76 @@ | ||
import datetime | ||
import time | ||
import uuid | ||
from typing import Any, Dict | ||
|
||
from aws_lambda_powertools.utilities.data_classes import AppSyncResolverEvent | ||
from aws_lambda_powertools.utilities.typing import LambdaContext | ||
|
||
|
||
def make_id(): | ||
return str(uuid.uuid4()) | ||
|
||
|
||
def aws_date(): | ||
now = datetime.datetime.utcnow().date() | ||
return now.strftime("%Y-%m-%d") | ||
|
||
|
||
def aws_time(): | ||
now = datetime.datetime.utcnow().time() | ||
return now.strftime("%H:%M:%S") | ||
|
||
|
||
def aws_datetime(): | ||
now = datetime.datetime.utcnow() | ||
return now.strftime("%Y-%m-%dT%H:%M:%SZ") | ||
|
||
|
||
def aws_timestamp(): | ||
return int(time.time()) | ||
|
||
|
||
class AppSyncResolver: | ||
def __init__(self): | ||
self._resolvers: dict = {} | ||
|
||
def resolver( | ||
self, | ||
type_name: str = "*", | ||
field_name: str = None, | ||
include_event: bool = False, | ||
include_context: bool = False, | ||
**kwargs, | ||
): | ||
def register_resolver(func): | ||
kwargs["include_event"] = include_event | ||
kwargs["include_context"] = include_context | ||
self._resolvers[f"{type_name}.{field_name}"] = { | ||
"func": func, | ||
"config": kwargs, | ||
} | ||
return func | ||
|
||
return register_resolver | ||
|
||
def resolve(self, event: dict, context: LambdaContext) -> Any: | ||
event = AppSyncResolverEvent(event) | ||
resolver, config = self._resolver(event.type_name, event.field_name) | ||
kwargs = self._kwargs(event, context, config) | ||
return resolver(**kwargs) | ||
|
||
def _resolver(self, type_name: str, field_name: str) -> tuple: | ||
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"], resolver["config"] | ||
|
||
@staticmethod | ||
def _kwargs(event: AppSyncResolverEvent, context: LambdaContext, config: dict) -> Dict[str, Any]: | ||
kwargs = {**event.arguments} | ||
if config.get("include_event", False): | ||
kwargs["event"] = event | ||
if config.get("include_context", False): | ||
kwargs["context"] = context | ||
return kwargs |
232 changes: 232 additions & 0 deletions
232
aws_lambda_powertools/utilities/data_classes/appsync_resolver_event.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,232 @@ | ||
from typing import Any, Dict, List, Optional, Union | ||
|
||
from aws_lambda_powertools.utilities.data_classes.common import DictWrapper, get_header_value | ||
|
||
|
||
def get_identity_object(identity: Optional[dict]) -> Any: | ||
"""Get the identity object based on the best detected type""" | ||
# API_KEY authorization | ||
if identity is None: | ||
return None | ||
|
||
# AMAZON_COGNITO_USER_POOLS authorization | ||
if "sub" in identity: | ||
return AppSyncIdentityCognito(identity) | ||
|
||
# AWS_IAM authorization | ||
return AppSyncIdentityIAM(identity) | ||
|
||
|
||
class AppSyncIdentityIAM(DictWrapper): | ||
"""AWS_IAM authorization""" | ||
|
||
@property | ||
def source_ip(self) -> List[str]: | ||
"""The source IP address of the caller received by AWS AppSync. """ | ||
return self["sourceIp"] | ||
|
||
@property | ||
def username(self) -> str: | ||
"""The user name of the authenticated user. IAM user principal""" | ||
return self["username"] | ||
|
||
@property | ||
def account_id(self) -> str: | ||
"""The AWS account ID of the caller.""" | ||
return self["accountId"] | ||
|
||
@property | ||
def cognito_identity_pool_id(self) -> str: | ||
"""The Amazon Cognito identity pool ID associated with the caller.""" | ||
return self["cognitoIdentityPoolId"] | ||
|
||
@property | ||
def cognito_identity_id(self) -> str: | ||
"""The Amazon Cognito identity ID of the caller.""" | ||
return self["cognitoIdentityId"] | ||
|
||
@property | ||
def user_arn(self) -> str: | ||
"""The ARN of the IAM user.""" | ||
return self["userArn"] | ||
|
||
@property | ||
def cognito_identity_auth_type(self) -> str: | ||
"""Either authenticated or unauthenticated based on the identity type.""" | ||
return self["cognitoIdentityAuthType"] | ||
|
||
@property | ||
def cognito_identity_auth_provider(self) -> str: | ||
"""A comma separated list of external identity provider information used in obtaining the | ||
credentials used to sign the request.""" | ||
return self["cognitoIdentityAuthProvider"] | ||
|
||
|
||
class AppSyncIdentityCognito(DictWrapper): | ||
"""AMAZON_COGNITO_USER_POOLS authorization""" | ||
|
||
@property | ||
def source_ip(self) -> List[str]: | ||
"""The source IP address of the caller received by AWS AppSync. """ | ||
return self["sourceIp"] | ||
|
||
@property | ||
def username(self) -> str: | ||
"""The user name of the authenticated user.""" | ||
return self["username"] | ||
|
||
@property | ||
def sub(self) -> str: | ||
"""The UUID of the authenticated user.""" | ||
return self["sub"] | ||
|
||
@property | ||
def claims(self) -> Dict[str, str]: | ||
"""The claims that the user has.""" | ||
return self["claims"] | ||
|
||
@property | ||
def default_auth_strategy(self) -> str: | ||
"""The default authorization strategy for this caller (ALLOW or DENY).""" | ||
return self["defaultAuthStrategy"] | ||
|
||
@property | ||
def groups(self) -> List[str]: | ||
"""List of OIDC groups""" | ||
return self["groups"] | ||
|
||
@property | ||
def issuer(self) -> str: | ||
"""The token issuer.""" | ||
return self["issuer"] | ||
|
||
|
||
class AppSyncResolverEventInfo(DictWrapper): | ||
"""The info section contains information about the GraphQL request""" | ||
|
||
@property | ||
def field_name(self) -> str: | ||
"""The name of the field that is currently being resolved.""" | ||
return self["fieldName"] | ||
|
||
@property | ||
def parent_type_name(self) -> str: | ||
"""The name of the parent type for the field that is currently being resolved.""" | ||
return self["parentTypeName"] | ||
|
||
@property | ||
def variables(self) -> Dict[str, str]: | ||
"""A map which holds all variables that are passed into the GraphQL request.""" | ||
return self.get("variables") | ||
|
||
@property | ||
def selection_set_list(self) -> List[str]: | ||
"""A list representation of the fields in the GraphQL selection set. Fields that are aliased will | ||
only be referenced by the alias name, not the field name.""" | ||
return self.get("selectionSetList") | ||
|
||
@property | ||
def selection_set_graphql(self) -> Optional[str]: | ||
"""A string representation of the selection set, formatted as GraphQL schema definition language (SDL). | ||
Although fragments are not be merged into the selection set, inline fragments are preserved.""" | ||
return self.get("selectionSetGraphQL") | ||
|
||
|
||
class AppSyncResolverEvent(DictWrapper): | ||
"""AppSync resolver event | ||
**NOTE:** AppSync Resolver Events can come in various shapes this data class | ||
supports both Amplify GraphQL directive @function and Direct Lambda Resolver | ||
Documentation: | ||
------------- | ||
- https://docs.aws.amazon.com/appsync/latest/devguide/resolver-context-reference.html | ||
- https://docs.amplify.aws/cli/graphql-transformer/function#structure-of-the-function-event | ||
""" | ||
|
||
def __init__(self, data: dict): | ||
super().__init__(data) | ||
|
||
info: dict = data.get("info") | ||
if not info: | ||
info = {"fieldName": self.get("fieldName"), "parentTypeName": self.get("typeName")} | ||
|
||
self._info = AppSyncResolverEventInfo(info) | ||
|
||
@property | ||
def type_name(self) -> str: | ||
"""The name of the parent type for the field that is currently being resolved.""" | ||
return self.info.parent_type_name | ||
|
||
@property | ||
def field_name(self) -> str: | ||
"""The name of the field that is currently being resolved.""" | ||
return self.info.field_name | ||
|
||
@property | ||
def arguments(self) -> Dict[str, any]: | ||
"""A map that contains all GraphQL arguments for this field.""" | ||
return self["arguments"] | ||
|
||
@property | ||
def identity(self) -> Union[None, AppSyncIdentityIAM, AppSyncIdentityCognito]: | ||
"""An object that contains information about the caller. | ||
Depending of the type of identify found: | ||
- API_KEY authorization - returns None | ||
- AWS_IAM authorization - returns AppSyncIdentityIAM | ||
- AMAZON_COGNITO_USER_POOLS authorization - returns AppSyncIdentityCognito | ||
""" | ||
return get_identity_object(self.get("identity")) | ||
|
||
@property | ||
def source(self) -> Dict[str, any]: | ||
"""A map that contains the resolution of the parent field.""" | ||
return self.get("source") | ||
|
||
@property | ||
def request_headers(self) -> Dict[str, str]: | ||
"""Request headers""" | ||
return self["request"]["headers"] | ||
|
||
@property | ||
def prev_result(self) -> Optional[Dict[str, any]]: | ||
"""It represents the result of whatever previous operation was executed in a pipeline resolver.""" | ||
prev = self.get("prev") | ||
if not prev: | ||
return None | ||
return prev.get("result") | ||
|
||
@property | ||
def info(self) -> AppSyncResolverEventInfo: | ||
"""The info section contains information about the GraphQL request.""" | ||
return self._info | ||
|
||
@property | ||
def stash(self) -> Optional[dict]: | ||
"""The stash is a map that is made available inside each resolver and function mapping template. | ||
The same stash instance lives through a single resolver execution. This means that you can use the | ||
stash to pass arbitrary data across request and response mapping templates, and across functions in | ||
a pipeline resolver.""" | ||
return self.get("stash") | ||
|
||
def get_header_value( | ||
self, name: str, default_value: Optional[str] = None, case_sensitive: Optional[bool] = False | ||
) -> Optional[str]: | ||
"""Get header value by name | ||
Parameters | ||
---------- | ||
name: str | ||
Header name | ||
default_value: str, optional | ||
Default value if no value was found by name | ||
case_sensitive: bool | ||
Whether to use a case sensitive look up | ||
Returns | ||
------- | ||
str, optional | ||
Header value | ||
""" | ||
return get_header_value(self.request_headers, name, default_value, case_sensitive) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.