Merge branch 'develop' into data_classes_duplicated_code

Author: leandrodamascena

Makefile

@@ -23,7 +23,7 @@ dev-gitpod:
 
 # Running licensecheck with zero to break the pipeline if there is an invalid license
 check-licenses:
-	poetry run licensecheck -u poetry:dev --zero
+	poetry run licensecheck -u poetry:dev
 
 format:
 	poetry run black aws_lambda_powertools tests examples

aws_lambda_powertools/metrics/provider/datadog/datadog.py

@@ -14,7 +14,7 @@
 from aws_lambda_powertools.metrics.provider import BaseProvider
 from aws_lambda_powertools.metrics.provider.datadog.warnings import DatadogDataValidationWarning
 from aws_lambda_powertools.shared import constants
-from aws_lambda_powertools.shared.functions import resolve_env_var_choice
+from aws_lambda_powertools.shared.functions import resolve_env_var_choice, strtobool
 
 if TYPE_CHECKING:
     from aws_lambda_powertools.shared.types import AnyCallableT
@@ -66,6 +66,9 @@ def __init__(
         )
         self.default_tags = default_tags or {}
         self.flush_to_log = resolve_env_var_choice(choice=flush_to_log, env=os.getenv(constants.DATADOG_FLUSH_TO_LOG))
+        # When set as env var, the value is a string
+        if isinstance(self.flush_to_log, str):
+            self.flush_to_log = strtobool(self.flush_to_log)
 
     #  adding name,value,timestamp,tags
     def add_metric(

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -4,6 +4,7 @@
 
 from .alb_event import ALBEvent
 from .api_gateway_proxy_event import APIGatewayProxyEvent, APIGatewayProxyEventV2
+from .api_gateway_websocket_event import APIGatewayWebSocketEvent
 from .appsync_resolver_event import AppSyncResolverEvent
 from .aws_config_rule_event import AWSConfigRuleEvent
 from .bedrock_agent_event import BedrockAgentEvent
@@ -51,6 +52,7 @@
 __all__ = [
     "APIGatewayProxyEvent",
     "APIGatewayProxyEventV2",
+    "APIGatewayWebSocketEvent",
     "SecretsManagerEvent",
     "AppSyncResolverEvent",
     "ALBEvent",

aws_lambda_powertools/utilities/data_classes/api_gateway_websocket_event.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+import base64
+from functools import cached_property
+from typing import Any
+
+from aws_lambda_powertools.utilities.data_classes.common import (
+    CaseInsensitiveDict,
+    DictWrapper,
+)
+
+
+class APIGatewayWebSocketEventIdentity(DictWrapper):
+    @property
+    def source_ip(self) -> str:
+        return self["sourceIp"]
+
+    @property
+    def user_agent(self) -> str | None:
+        return self.get("userAgent")
+
+
+class APIGatewayWebSocketEventRequestContext(DictWrapper):
+    @property
+    def route_key(self) -> str:
+        return self["routeKey"]
+
+    @property
+    def disconnect_status_code(self) -> int | None:
+        return self.get("disconnectStatusCode")
+
+    @property
+    def message_id(self) -> str | None:
+        return self.get("messageId")
+
+    @property
+    def event_type(self) -> str:
+        return self["eventType"]
+
+    @property
+    def extended_request_id(self) -> str:
+        return self["extendedRequestId"]
+
+    @property
+    def request_time(self) -> str:
+        return self["requestTime"]
+
+    @property
+    def message_direction(self) -> str:
+        return self["messageDirection"]
+
+    @property
+    def disconnect_reason(self) -> str | None:
+        return self.get("disconnectReason")
+
+    @property
+    def stage(self) -> str:
+        return self["stage"]
+
+    @property
+    def connected_at(self) -> int:
+        return self["connectedAt"]
+
+    @property
+    def request_time_epoch(self) -> int:
+        return self["requestTimeEpoch"]
+
+    @property
+    def identity(self) -> APIGatewayWebSocketEventIdentity:
+        return APIGatewayWebSocketEventIdentity(self["identity"])
+
+    @property
+    def request_id(self) -> str:
+        return self["requestId"]
+
+    @property
+    def domain_name(self) -> str:
+        return self["domainName"]
+
+    @property
+    def connection_id(self) -> str:
+        return self["connectionId"]
+
+    @property
+    def api_id(self) -> str:
+        return self["apiId"]
+
+
+class APIGatewayWebSocketEvent(DictWrapper):
+    """AWS proxy integration event for WebSocket API
+
+    Documentation:
+    --------------
+    - https://docs.aws.amazon.com/apigateway/latest/developerguide/apigateway-websocket-api-integration-requests.html
+    """
+
+    @property
+    def is_base64_encoded(self) -> bool:
+        return self["isBase64Encoded"]
+
+    @property
+    def body(self) -> str | None:
+        return self.get("body")
+
+    @cached_property
+    def decoded_body(self) -> str | None:
+        body = self.body
+        if self.is_base64_encoded and body:
+            return base64.b64decode(body.encode()).decode()
+        return body
+
+    @cached_property
+    def json_body(self) -> Any:
+        if self.decoded_body:
+            return self._json_deserializer(self.decoded_body)
+        return None
+
+    @property
+    def headers(self) -> dict[str, str]:
+        return CaseInsensitiveDict(self.get("headers"))
+
+    @property
+    def multi_value_headers(self) -> dict[str, list[str]]:
+        return CaseInsensitiveDict(self.get("multiValueHeaders"))
+
+    @property
+    def request_context(self) -> APIGatewayWebSocketEventRequestContext:
+        return APIGatewayWebSocketEventRequestContext(self["requestContext"])

noxfile.py

@@ -80,6 +80,7 @@ def test_with_datadog_as_required_package(session: nox.Session):
         folders=[
             f"{PREFIX_TESTS_FUNCTIONAL}/metrics/datadog/",
         ],
+        extras="datadog,aws-sdk",  # Datadog library requires boto3
     )