Module aws_lambda_powertools.utilities.data_classes.common
Expand source code
from typing import Any, Dict, Optional
class DictWrapper:
"""Provides a single read only access to a wrapper dict"""
def __init__(self, data: Dict[str, Any]):
self._data = data
def __getitem__(self, key: str) -> Any:
return self._data[key]
def __eq__(self, other: Any) -> bool:
if not isinstance(other, DictWrapper):
return False
return self._data == other._data
def get(self, key: str) -> Optional[Any]:
return self._data.get(key)
@property
def raw_event(self) -> Dict[str, Any]:
"""The original raw event dict"""
return self._data
def get_header_value(
headers: Dict[str, str], name: str, default_value: Optional[str], case_sensitive: Optional[bool]
) -> Optional[str]:
"""Get header value by name"""
if case_sensitive:
return headers.get(name, default_value)
name_lower = name.lower()
return next(
# Iterate over the dict and do a case insensitive key comparison
(value for key, value in headers.items() if key.lower() == name_lower),
# Default value is returned if no matches was found
default_value,
)
class BaseProxyEvent(DictWrapper):
@property
def headers(self) -> Dict[str, str]:
return self["headers"]
@property
def query_string_parameters(self) -> Optional[Dict[str, str]]:
return self.get("queryStringParameters")
@property
def is_base64_encoded(self) -> Optional[bool]:
return self.get("isBase64Encoded")
@property
def body(self) -> Optional[str]:
return self.get("body")
def get_query_string_value(self, name: str, default_value: Optional[str] = None) -> Optional[str]:
"""Get query string value by name
Parameters
----------
name: str
Query string parameter name
default_value: str, optional
Default value if no value was found by name
Returns
-------
str, optional
Query string parameter value
"""
params = self.query_string_parameters
return default_value if params is None else params.get(name, default_value)
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.headers, name, default_value, case_sensitive)Functions
def get_header_value(headers: Dict[str, str], name: str, default_value: Union[str, NoneType], case_sensitive: Union[bool, NoneType]) ‑> Union[str, NoneType]-
Get header value by name
Expand source code
def get_header_value( headers: Dict[str, str], name: str, default_value: Optional[str], case_sensitive: Optional[bool] ) -> Optional[str]: """Get header value by name""" if case_sensitive: return headers.get(name, default_value) name_lower = name.lower() return next( # Iterate over the dict and do a case insensitive key comparison (value for key, value in headers.items() if key.lower() == name_lower), # Default value is returned if no matches was found default_value, )
Classes
class BaseProxyEvent (data: Dict[str, Any])-
Provides a single read only access to a wrapper dict
Expand source code
class BaseProxyEvent(DictWrapper): @property def headers(self) -> Dict[str, str]: return self["headers"] @property def query_string_parameters(self) -> Optional[Dict[str, str]]: return self.get("queryStringParameters") @property def is_base64_encoded(self) -> Optional[bool]: return self.get("isBase64Encoded") @property def body(self) -> Optional[str]: return self.get("body") def get_query_string_value(self, name: str, default_value: Optional[str] = None) -> Optional[str]: """Get query string value by name Parameters ---------- name: str Query string parameter name default_value: str, optional Default value if no value was found by name Returns ------- str, optional Query string parameter value """ params = self.query_string_parameters return default_value if params is None else params.get(name, default_value) 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.headers, name, default_value, case_sensitive)Ancestors
Subclasses
Instance variables
var body : Union[str, NoneType]-
Expand source code
@property def body(self) -> Optional[str]: return self.get("body") var headers : Dict[str, str]-
Expand source code
@property def headers(self) -> Dict[str, str]: return self["headers"] var is_base64_encoded : Union[bool, NoneType]-
Expand source code
@property def is_base64_encoded(self) -> Optional[bool]: return self.get("isBase64Encoded") var query_string_parameters : Union[Dict[str, str], NoneType]-
Expand source code
@property def query_string_parameters(self) -> Optional[Dict[str, str]]: return self.get("queryStringParameters")
Methods
def get_header_value(self, name: str, default_value: Union[str, NoneType] = None, case_sensitive: Union[bool, NoneType] = False) ‑> Union[str, NoneType]-
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
Expand source code
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.headers, name, default_value, case_sensitive) def get_query_string_value(self, name: str, default_value: Union[str, NoneType] = None) ‑> Union[str, NoneType]-
Get query string value by name
Parameters
name:str- Query string parameter name
default_value:str, optional- Default value if no value was found by name
Returns
str, optional- Query string parameter value
Expand source code
def get_query_string_value(self, name: str, default_value: Optional[str] = None) -> Optional[str]: """Get query string value by name Parameters ---------- name: str Query string parameter name default_value: str, optional Default value if no value was found by name Returns ------- str, optional Query string parameter value """ params = self.query_string_parameters return default_value if params is None else params.get(name, default_value)
Inherited members
class DictWrapper (data: Dict[str, Any])-
Provides a single read only access to a wrapper dict
Expand source code
class DictWrapper: """Provides a single read only access to a wrapper dict""" def __init__(self, data: Dict[str, Any]): self._data = data def __getitem__(self, key: str) -> Any: return self._data[key] def __eq__(self, other: Any) -> bool: if not isinstance(other, DictWrapper): return False return self._data == other._data def get(self, key: str) -> Optional[Any]: return self._data.get(key) @property def raw_event(self) -> Dict[str, Any]: """The original raw event dict""" return self._dataSubclasses
- ALBEventRequestContext
- APIGatewayEventAuthorizer
- APIGatewayEventIdentity
- APIGatewayEventRequestContext
- RequestContextV2
- RequestContextV2Authorizer
- RequestContextV2AuthorizerIam
- RequestContextV2Http
- AppSyncIdentityCognito
- AppSyncIdentityIAM
- AppSyncResolverEvent
- AppSyncResolverEventInfo
- CloudWatchLogsDecodedData
- CloudWatchLogsEvent
- CloudWatchLogsLogEvent
- BaseTriggerEvent
- CallerContext
- ChallengeResult
- ClaimsOverrideDetails
- CreateAuthChallengeTriggerEventRequest
- CreateAuthChallengeTriggerEventResponse
- CustomMessageTriggerEventRequest
- CustomMessageTriggerEventResponse
- DefineAuthChallengeTriggerEventRequest
- DefineAuthChallengeTriggerEventResponse
- GroupOverrideDetails
- PostAuthenticationTriggerEventRequest
- PostConfirmationTriggerEventRequest
- PreAuthenticationTriggerEventRequest
- PreSignUpTriggerEventRequest
- PreSignUpTriggerEventResponse
- PreTokenGenerationTriggerEventRequest
- PreTokenGenerationTriggerEventResponse
- UserMigrationTriggerEventRequest
- UserMigrationTriggerEventResponse
- VerifyAuthChallengeResponseTriggerEventRequest
- VerifyAuthChallengeResponseTriggerEventResponse
- BaseProxyEvent
- ConnectContactFlowData
- ConnectContactFlowEndpoint
- ConnectContactFlowEvent
- ConnectContactFlowMediaStreamAudio
- ConnectContactFlowMediaStreamCustomer
- ConnectContactFlowMediaStreams
- ConnectContactFlowQueue
- AttributeValue
- DynamoDBRecord
- DynamoDBStreamEvent
- StreamRecord
- EventBridgeEvent
- KinesisStreamEvent
- KinesisStreamRecord
- KinesisStreamRecordPayload
- S3Bucket
- S3Event
- S3EventRecord
- S3EventRecordGlacierEventData
- S3EventRecordGlacierRestoreEventData
- S3Identity
- S3Message
- S3Object
- S3RequestParameters
- S3ObjectConfiguration
- S3ObjectContext
- S3ObjectLambdaEvent
- S3ObjectSessionAttributes
- S3ObjectSessionContext
- S3ObjectSessionIssuer
- S3ObjectUserIdentity
- S3ObjectUserRequest
- SESEvent
- SESEventRecord
- SESMail
- SESMailCommonHeaders
- SESMailHeader
- SESMessage
- SESReceipt
- SESReceiptAction
- SESReceiptStatus
- SNSEvent
- SNSEventRecord
- SNSMessage
- SNSMessageAttribute
- SQSEvent
- SQSMessageAttribute
- SQSRecord
- SQSRecordAttributes
Instance variables
var raw_event : Dict[str, Any]-
The original raw event dict
Expand source code
@property def raw_event(self) -> Dict[str, Any]: """The original raw event dict""" return self._data
Methods
def get(self, key: str) ‑> Union[Any, NoneType]-
Expand source code
def get(self, key: str) -> Optional[Any]: return self._data.get(key)