Skip to content

feat(trigger): data class and helper functions for lambda trigger events #159

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 30 commits into from
Sep 18, 2020
Merged

feat(trigger): data class and helper functions for lambda trigger events #159

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
30 commits
Select commit Hold shift + click to select a range
a3e3019
feat(trigger): class wrapper for events
michaelbrewer Sep 7, 2020
2c50acb
build: fix build for python 3.6
michaelbrewer Sep 7, 2020
bdf4385
fix(linters): make the python linters happy
michaelbrewer Sep 7, 2020
112635d
tests: add missing test cases
michaelbrewer Sep 7, 2020
d9253f5
docs: Include some docstrings
michaelbrewer Sep 7, 2020
057c947
feat(trigger): initial cognito triggers
michaelbrewer Sep 8, 2020
84b63a3
feat(trigger): add event_bridge_event
michaelbrewer Sep 8, 2020
c1d516b
feat(trigger): use consistent getter name
michaelbrewer Sep 8, 2020
2aea86c
refactor: less copies passed around
michaelbrewer Sep 8, 2020
17906ab
feat(trigger): add UserMigrationTriggerEvent
michaelbrewer Sep 8, 2020
1c69762
feat(trigger): cognito custom message and pre-auth
michaelbrewer Sep 9, 2020
b9cdd66
feat(trigger): cognito pre token and post auth
michaelbrewer Sep 9, 2020
87277ba
tests(trigger): some extra checks
michaelbrewer Sep 9, 2020
92a3ca7
chore(trigger): clean up and docs
michaelbrewer Sep 9, 2020
f391461
chore: consistent naming
michaelbrewer Sep 10, 2020
b396e8c
feat(trigger): Add api gateway proxy events
michaelbrewer Sep 10, 2020
553c48a
chore: Add more docs
michaelbrewer Sep 10, 2020
7e6a3e5
fix(trigger): better type hinting
michaelbrewer Sep 12, 2020
7a7e862
feat(trigger): Add conveince methods
michaelbrewer Sep 12, 2020
379b11f
feat(trigger): Create DictWrapper
michaelbrewer Sep 13, 2020
cc180d3
feat(trigger): API gateway helper methods
michaelbrewer Sep 13, 2020
3811356
feat(trigger): Kinesis stream event
michaelbrewer Sep 13, 2020
3c41f36
feat(trigger): Application load balancer event
michaelbrewer Sep 14, 2020
35358ed
refactor(trigger): Split decompress and parse
michaelbrewer Sep 14, 2020
a746ddc
feat(trigger): Some attribte caching
michaelbrewer Sep 14, 2020
7b4ec44
fix(trigger): Init with __init__
michaelbrewer Sep 14, 2020
6306436
fix(trigger): Add missing __eq__
michaelbrewer Sep 14, 2020
0ffb1be
feat(trigger): unquote_plus s3 object key
michaelbrewer Sep 14, 2020
d0c9e5b
refactor(data_classes): rename package from trigger
michaelbrewer Sep 17, 2020
e860731
refactor(data_classes): Use DictWrapper consistently
michaelbrewer Sep 17, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
134 changes: 67 additions & 67 deletions aws_lambda_powertools/utilities/trigger/api_gateway_proxy_event.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,205 +2,205 @@


class APIGatewayEventIdentity:
def __init__(self, event: dict):
self._val = event
def __init__(self, event: Dict[str, Any]):
self._v = event

@property
def access_key(self) -> Optional[str]:
return self._val["requestContext"]["identity"].get("accessKey")
return self._v["requestContext"]["identity"].get("accessKey")

@property
def account_id(self) -> Optional[str]:
"""The AWS account ID associated with the request."""
return self._val["requestContext"]["identity"].get("accountId")
return self._v["requestContext"]["identity"].get("accountId")

@property
def api_key(self) -> Optional[str]:
"""For API methods that require an API key, this variable is the API key associated with the method request.
For methods that don't require an API key, this variable is null. """
return self._val["requestContext"]["identity"].get("apiKey")
return self._v["requestContext"]["identity"].get("apiKey")

@property
def api_key_id(self) -> Optional[str]:
"""The API key ID associated with an API request that requires an API key."""
return self._val["requestContext"]["identity"].get("apiKeyId")
return self._v["requestContext"]["identity"].get("apiKeyId")

@property
def caller(self) -> Optional[str]:
"""The principal identifier of the caller making the request."""
return self._val["requestContext"]["identity"].get("caller")
return self._v["requestContext"]["identity"].get("caller")

@property
def cognito_authentication_provider(self) -> Optional[str]:
"""A comma-separated list of the Amazon Cognito authentication providers used by the caller
making the request. Available only if the request was signed with Amazon Cognito credentials."""
return self._val["requestContext"]["identity"].get("cognitoAuthenticationProvider")
return self._v["requestContext"]["identity"].get("cognitoAuthenticationProvider")

@property
def cognito_authentication_type(self) -> Optional[str]:
"""The Amazon Cognito authentication type of the caller making the request.
Available only if the request was signed with Amazon Cognito credentials."""
return self._val["requestContext"]["identity"].get("cognitoAuthenticationType")
return self._v["requestContext"]["identity"].get("cognitoAuthenticationType")

@property
def cognito_identity_id(self) -> Optional[str]:
"""The Amazon Cognito identity ID of the caller making the request.
Available only if the request was signed with Amazon Cognito credentials."""
return self._val["requestContext"]["identity"].get("cognitoIdentityId")
return self._v["requestContext"]["identity"].get("cognitoIdentityId")

@property
def cognito_identity_pool_id(self) -> Optional[str]:
"""The Amazon Cognito identity pool ID of the caller making the request.
Available only if the request was signed with Amazon Cognito credentials."""
return self._val["requestContext"]["identity"].get("cognitoIdentityPoolId")
return self._v["requestContext"]["identity"].get("cognitoIdentityPoolId")

@property
def principal_org_id(self) -> Optional[str]:
"""The AWS organization ID."""
return self._val["requestContext"]["identity"].get("principalOrgId")
return self._v["requestContext"]["identity"].get("principalOrgId")

@property
def source_ip(self) -> str:
"""The source IP address of the TCP connection making the request to API Gateway."""
return self._val["requestContext"]["identity"]["sourceIp"]
return self._v["requestContext"]["identity"]["sourceIp"]

@property
def user(self) -> Optional[str]:
"""The principal identifier of the user making the request."""
return self._val["requestContext"]["identity"].get("user")
return self._v["requestContext"]["identity"].get("user")

@property
def user_agent(self) -> Optional[str]:
"""The User Agent of the API caller."""
return self._val["requestContext"]["identity"].get("userAgent")
return self._v["requestContext"]["identity"].get("userAgent")

@property
def user_arn(self) -> Optional[str]:
"""The Amazon Resource Name (ARN) of the effective user identified after authentication."""
return self._val["requestContext"]["identity"].get("userArn")
return self._v["requestContext"]["identity"].get("userArn")


class APIGatewayEventAuthorizer:
def __init__(self, event: Dict):
self._val = event
def __init__(self, event: Dict[str, Any]):
self._v = event

@property
def claims(self) -> Optional[Dict[str, Any]]:
return self._val["requestContext"]["authorizer"].get("claims")
return self._v["requestContext"]["authorizer"].get("claims")

@property
def scopes(self) -> Optional[List[str]]:
return self._val["requestContext"]["authorizer"].get("scopes")
return self._v["requestContext"]["authorizer"].get("scopes")


class APIGatewayEventRequestContext:
def __init__(self, event: Dict[str, Any]):
self._val = event
self._v = event

@property
def account_id(self) -> str:
"""The AWS account ID associated with the request."""
return self._val["requestContext"]["accountId"]
return self._v["requestContext"]["accountId"]

@property
def api_id(self) -> str:
"""The identifier API Gateway assigns to your API."""
return self._val["requestContext"]["apiId"]
return self._v["requestContext"]["apiId"]

@property
def authorizer(self) -> APIGatewayEventAuthorizer:
return APIGatewayEventAuthorizer(self._val)
return APIGatewayEventAuthorizer(self._v)

@property
def connected_at(self) -> Optional[int]:
"""The Epoch-formatted connection time. (WebSocket API)"""
return self._val["requestContext"].get("connectedAt")
return self._v["requestContext"].get("connectedAt")

@property
def connection_id(self) -> Optional[str]:
"""A unique ID for the connection that can be used to make a callback to the client. (WebSocket API)"""
return self._val["requestContext"].get("connectionId")
return self._v["requestContext"].get("connectionId")

@property
def domain_name(self) -> Optional[str]:
"""A domain name"""
return self._val["requestContext"].get("domainName")
return self._v["requestContext"].get("domainName")

@property
def domain_prefix(self) -> Optional[str]:
return self._val["requestContext"].get("domainPrefix")
return self._v["requestContext"].get("domainPrefix")

@property
def event_type(self) -> Optional[str]:
"""The event type: `CONNECT`, `MESSAGE`, or `DISCONNECT`. (WebSocket API)"""
return self._val["requestContext"].get("eventType")
return self._v["requestContext"].get("eventType")

@property
def extended_request_id(self) -> Optional[str]:
"""An automatically generated ID for the API call, which contains more useful information
for debugging/troubleshooting."""
return self._val["requestContext"].get("extendedRequestId")
return self._v["requestContext"].get("extendedRequestId")

@property
def protocol(self) -> str:
"""The request protocol, for example, HTTP/1.1."""
return self._val["requestContext"]["protocol"]
return self._v["requestContext"]["protocol"]

@property
def http_method(self) -> str:
"""The HTTP method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT."""
return self._val["requestContext"]["httpMethod"]
return self._v["requestContext"]["httpMethod"]

@property
def identity(self) -> APIGatewayEventIdentity:
return APIGatewayEventIdentity(self._val)
return APIGatewayEventIdentity(self._v)

@property
def message_direction(self) -> Optional[str]:
"""Message direction (WebSocket API)"""
return self._val["requestContext"].get("messageDirection")
return self._v["requestContext"].get("messageDirection")

@property
def message_id(self) -> Optional[str]:
"""A unique server-side ID for a message. Available only when the `eventType` is `MESSAGE`."""
return self._val["requestContext"].get("messageId")
return self._v["requestContext"].get("messageId")

@property
def path(self) -> str:
return self._val["requestContext"]["path"]
return self._v["requestContext"]["path"]

@property
def stage(self) -> str:
"""The deployment stage of the API request """
return self._val["requestContext"]["stage"]
return self._v["requestContext"]["stage"]

@property
def request_id(self) -> str:
"""The ID that API Gateway assigns to the API request."""
return self._val["requestContext"]["requestId"]
return self._v["requestContext"]["requestId"]

@property
def request_time(self) -> Optional[str]:
"""The CLF-formatted request time (dd/MMM/yyyy:HH:mm:ss +-hhmm)"""
return self._val["requestContext"].get("requestTime")
return self._v["requestContext"].get("requestTime")

@property
def request_time_epoch(self) -> int:
"""The Epoch-formatted request time."""
return self._val["requestContext"]["requestTimeEpoch"]
return self._v["requestContext"]["requestTimeEpoch"]

@property
def resource_id(self) -> str:
return self._val["requestContext"]["resourceId"]
return self._v["requestContext"]["resourceId"]

@property
def resource_path(self) -> str:
return self._val["requestContext"]["resourcePath"]
return self._v["requestContext"]["resourcePath"]

@property
def route_key(self) -> Optional[str]:
"""The selected route key."""
return self._val["requestContext"].get("routeKey")
return self._v["requestContext"].get("routeKey")


class APIGatewayProxyEvent(dict):
Expand Down Expand Up @@ -266,102 +266,102 @@ def is_base64_encoded(self) -> bool:


class RequestContextV2Http:
def __init__(self, event: dict):
self._val = event
def __init__(self, event: Dict[str, Any]):
self._v = event

@property
def method(self) -> str:
return self._val["requestContext"]["http"]["method"]
return self._v["requestContext"]["http"]["method"]

@property
def path(self) -> str:
return self._val["requestContext"]["http"]["path"]
return self._v["requestContext"]["http"]["path"]

@property
def protocol(self) -> str:
"""The request protocol, for example, HTTP/1.1."""
return self._val["requestContext"]["http"]["protocol"]
return self._v["requestContext"]["http"]["protocol"]

@property
def source_ip(self) -> str:
"""The source IP address of the TCP connection making the request to API Gateway."""
return self._val["requestContext"]["http"]["sourceIp"]
return self._v["requestContext"]["http"]["sourceIp"]

@property
def user_agent(self) -> str:
"""The User Agent of the API caller."""
return self._val["requestContext"]["http"]["userAgent"]
return self._v["requestContext"]["http"]["userAgent"]


class RequestContextV2Authorizer:
def __init__(self, event: dict):
self._val = event
def __init__(self, event: Dict[str, Any]):
self._v = event

@property
def jwt_claim(self) -> Dict[str, Any]:
return self._val["jwt"]["claims"]
return self._v["jwt"]["claims"]

@property
def jwt_scopes(self) -> List[str]:
return self._val["jwt"]["scopes"]
return self._v["jwt"]["scopes"]


class RequestContextV2:
def __init__(self, event: dict):
self._val = event
def __init__(self, event: Dict[str, Any]):
self._v = event

@property
def account_id(self) -> str:
"""The AWS account ID associated with the request."""
return self._val["requestContext"]["accountId"]
return self._v["requestContext"]["accountId"]

@property
def api_id(self) -> str:
"""The identifier API Gateway assigns to your API."""
return self._val["requestContext"]["apiId"]
return self._v["requestContext"]["apiId"]

@property
def authorizer(self) -> Optional[RequestContextV2Authorizer]:
authorizer = self._val["requestContext"].get("authorizer")
authorizer = self._v["requestContext"].get("authorizer")
return None if authorizer is None else RequestContextV2Authorizer(authorizer)

@property
def domain_name(self) -> str:
"""A domain name """
return self._val["requestContext"]["domainName"]
return self._v["requestContext"]["domainName"]

@property
def domain_prefix(self) -> str:
return self._val["requestContext"]["domainPrefix"]
return self._v["requestContext"]["domainPrefix"]

@property
def http(self) -> RequestContextV2Http:
return RequestContextV2Http(self._val)
return RequestContextV2Http(self._v)

@property
def request_id(self) -> str:
"""The ID that API Gateway assigns to the API request."""
return self._val["requestContext"]["requestId"]
return self._v["requestContext"]["requestId"]

@property
def route_key(self) -> str:
"""The selected route key."""
return self._val["requestContext"]["routeKey"]
return self._v["requestContext"]["routeKey"]

@property
def stage(self) -> str:
"""The deployment stage of the API request """
return self._val["requestContext"]["stage"]
return self._v["requestContext"]["stage"]

@property
def time(self) -> str:
"""The CLF-formatted request time (dd/MMM/yyyy:HH:mm:ss +-hhmm)."""
return self._val["requestContext"]["time"]
return self._v["requestContext"]["time"]

@property
def time_epoch(self) -> int:
"""The Epoch-formatted request time."""
return self._val["requestContext"]["timeEpoch"]
return self._v["requestContext"]["timeEpoch"]


class APIGatewayProxyEventV2(dict):
Expand Down
Loading