Module aws_lambda_powertools.utilities.data_classes.cloud_watch_logs_event

Classes

class CloudWatchLogsDecodedData (data: dict[str, Any], json_deserializer: Callable | None = None)

Provides a single read only access to a wrapper dict

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class CloudWatchLogsDecodedData(DictWrapper):
    @property
    def owner(self) -> str:
        """The AWS Account ID of the originating log data."""
        return self["owner"]

    @property
    def log_group(self) -> str:
        """The log group name of the originating log data."""
        return self["logGroup"]

    @property
    def log_stream(self) -> str:
        """The log stream name of the originating log data."""
        return self["logStream"]

    @property
    def subscription_filters(self) -> list[str]:
        """The list of subscription filter names that matched with the originating log data."""
        return self["subscriptionFilters"]

    @property
    def message_type(self) -> str:
        """Data messages will use the "DATA_MESSAGE" type.

        Sometimes CloudWatch Logs may emit Kinesis records with a "CONTROL_MESSAGE" type,
        mainly for checking if the destination is reachable.
        """
        return self["messageType"]

    @property
    def policy_level(self) -> str | None:
        """The level at which the policy was enforced."""
        return self.get("policyLevel")

    @property
    def log_events(self) -> list[CloudWatchLogsLogEvent]:
        """The actual log data, represented as an array of log event records.

        The ID property is a unique identifier for every log event.
        """
        return [CloudWatchLogsLogEvent(i) for i in self["logEvents"]]

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop log_events : list[CloudWatchLogsLogEvent]

The actual log data, represented as an array of log event records.

The ID property is a unique identifier for every log event.

Expand source code

@property
def log_events(self) -> list[CloudWatchLogsLogEvent]:
    """The actual log data, represented as an array of log event records.

    The ID property is a unique identifier for every log event.
    """
    return [CloudWatchLogsLogEvent(i) for i in self["logEvents"]]

prop log_group : str

The log group name of the originating log data.

Expand source code

@property
def log_group(self) -> str:
    """The log group name of the originating log data."""
    return self["logGroup"]

prop log_stream : str

The log stream name of the originating log data.

Expand source code

@property
def log_stream(self) -> str:
    """The log stream name of the originating log data."""
    return self["logStream"]

prop message_type : str

Data messages will use the "DATA_MESSAGE" type.

Sometimes CloudWatch Logs may emit Kinesis records with a "CONTROL_MESSAGE" type, mainly for checking if the destination is reachable.

Expand source code

@property
def message_type(self) -> str:
    """Data messages will use the "DATA_MESSAGE" type.

    Sometimes CloudWatch Logs may emit Kinesis records with a "CONTROL_MESSAGE" type,
    mainly for checking if the destination is reachable.
    """
    return self["messageType"]

prop owner : str

The AWS Account ID of the originating log data.

Expand source code

@property
def owner(self) -> str:
    """The AWS Account ID of the originating log data."""
    return self["owner"]

prop policy_level : str | None

The level at which the policy was enforced.

Expand source code

@property
def policy_level(self) -> str | None:
    """The level at which the policy was enforced."""
    return self.get("policyLevel")

prop subscription_filters : list[str]

The list of subscription filter names that matched with the originating log data.

Expand source code

@property
def subscription_filters(self) -> list[str]:
    """The list of subscription filter names that matched with the originating log data."""
    return self["subscriptionFilters"]

Inherited members

• DictWrapper:
  • get
  • raw_event

class CloudWatchLogsEvent (data: dict[str, Any], json_deserializer: Callable | None = None)

CloudWatch Logs log stream event

You can use a Lambda function to monitor and analyze logs from an Amazon CloudWatch Logs log stream.

Documentation:

• https://docs.aws.amazon.com/lambda/latest/dg/services-cloudwatchlogs.html

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class CloudWatchLogsEvent(DictWrapper):
    """CloudWatch Logs log stream event

    You can use a Lambda function to monitor and analyze logs from an Amazon CloudWatch Logs log stream.

    Documentation:
    --------------
    - https://docs.aws.amazon.com/lambda/latest/dg/services-cloudwatchlogs.html
    """

    _decompressed_logs_data = None
    _json_logs_data = None

    @property
    def raw_logs_data(self) -> str:
        """The value of the `data` field is a Base64 encoded ZIP archive."""
        return self["awslogs"]["data"]

    @property
    def decompress_logs_data(self) -> bytes:
        """Decode and decompress log data"""
        if self._decompressed_logs_data is None:
            payload = base64.b64decode(self.raw_logs_data)
            self._decompressed_logs_data = zlib.decompress(payload, zlib.MAX_WBITS | 32)
        return self._decompressed_logs_data

    def parse_logs_data(self) -> CloudWatchLogsDecodedData:
        """Decode, decompress and parse json data as CloudWatchLogsDecodedData"""
        if self._json_logs_data is None:
            self._json_logs_data = self._json_deserializer(self.decompress_logs_data.decode("UTF-8"))

        return CloudWatchLogsDecodedData(self._json_logs_data)

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop decompress_logs_data : bytes

Decode and decompress log data

Expand source code

@property
def decompress_logs_data(self) -> bytes:
    """Decode and decompress log data"""
    if self._decompressed_logs_data is None:
        payload = base64.b64decode(self.raw_logs_data)
        self._decompressed_logs_data = zlib.decompress(payload, zlib.MAX_WBITS | 32)
    return self._decompressed_logs_data

prop raw_logs_data : str

The value of the data field is a Base64 encoded ZIP archive.

Expand source code

@property
def raw_logs_data(self) -> str:
    """The value of the `data` field is a Base64 encoded ZIP archive."""
    return self["awslogs"]["data"]

Methods

def parse_logs_data(self) ‑> CloudWatchLogsDecodedData

Decode, decompress and parse json data as CloudWatchLogsDecodedData

Inherited members

• DictWrapper:
  • get
  • raw_event

class CloudWatchLogsLogEvent (data: dict[str, Any], json_deserializer: Callable | None = None)

Provides a single read only access to a wrapper dict

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class CloudWatchLogsLogEvent(DictWrapper):
    @property
    def get_id(self) -> str:
        """The ID property is a unique identifier for every log event."""
        # Note: this name conflicts with existing python builtins
        return self["id"]

    @property
    def timestamp(self) -> int:
        """Get the `timestamp` property"""
        return self["timestamp"]

    @property
    def message(self) -> str:
        """Get the `message` property"""
        return self["message"]

    @property
    def extracted_fields(self) -> dict[str, str]:
        """Get the `extractedFields` property"""
        return self.get("extractedFields") or {}

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop extracted_fields : dict[str, str]

Get the extractedFields property

Expand source code

@property
def extracted_fields(self) -> dict[str, str]:
    """Get the `extractedFields` property"""
    return self.get("extractedFields") or {}

prop get_id : str

The ID property is a unique identifier for every log event.

Expand source code

@property
def get_id(self) -> str:
    """The ID property is a unique identifier for every log event."""
    # Note: this name conflicts with existing python builtins
    return self["id"]

prop message : str

Get the message property

Expand source code

@property
def message(self) -> str:
    """Get the `message` property"""
    return self["message"]

prop timestamp : int

Get the timestamp property

Expand source code

@property
def timestamp(self) -> int:
    """Get the `timestamp` property"""
    return self["timestamp"]

Inherited members

• DictWrapper:
  • get
  • raw_event