Module aws_lambda_powertools.utilities.data_classes.kinesis_stream_event

class KinesisStreamEvent(DictWrapper):
    """Kinesis stream event

    Documentation:
    --------------
    - https://docs.aws.amazon.com/lambda/latest/dg/with-kinesis.html
    """

    @property
    def records(self) -> Iterator[KinesisStreamRecord]:
        for record in self["Records"]:
            yield KinesisStreamRecord(record)

Ancestors

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

Instance variables

prop records : Iterator[KinesisStreamRecord]

Expand source code

@property
def records(self) -> Iterator[KinesisStreamRecord]:
    for record in self["Records"]:
        yield KinesisStreamRecord(record)

Inherited members

• DictWrapper:
  • get
  • raw_event

class KinesisStreamRecord(DictWrapper):
    @property
    def aws_region(self) -> str:
        """AWS region where the event originated eg: us-east-1"""
        return self["awsRegion"]

    @property
    def event_id(self) -> str:
        """A globally unique identifier for the event that was recorded in this stream record."""
        return self["eventID"]

    @property
    def event_name(self) -> str:
        """Event type eg: aws:kinesis:record"""
        return self["eventName"]

    @property
    def event_source(self) -> str:
        """The AWS service from which the Kinesis event originated. For Kinesis, this is aws:kinesis"""
        return self["eventSource"]

    @property
    def event_source_arn(self) -> str:
        """The Amazon Resource Name (ARN) of the event source"""
        return self["eventSourceARN"]

    @property
    def event_version(self) -> str:
        """The eventVersion key value contains a major and minor version in the form <major>.<minor>."""
        return self["eventVersion"]

    @property
    def invoke_identity_arn(self) -> str:
        """The ARN for the identity used to invoke the Lambda Function"""
        return self["invokeIdentityArn"]

    @property
    def kinesis(self) -> KinesisStreamRecordPayload:
        """Underlying Kinesis record associated with the event"""
        return KinesisStreamRecordPayload(self._data)

Ancestors

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

Instance variables

prop aws_region : str

Expand source code

@property
def aws_region(self) -> str:
    """AWS region where the event originated eg: us-east-1"""
    return self["awsRegion"]

AWS region where the event originated eg: us-east-1

prop event_id : str

Expand source code

@property
def event_id(self) -> str:
    """A globally unique identifier for the event that was recorded in this stream record."""
    return self["eventID"]

A globally unique identifier for the event that was recorded in this stream record.

prop event_name : str

Expand source code

@property
def event_name(self) -> str:
    """Event type eg: aws:kinesis:record"""
    return self["eventName"]

Event type eg: aws:kinesis:record

prop event_source : str

Expand source code

@property
def event_source(self) -> str:
    """The AWS service from which the Kinesis event originated. For Kinesis, this is aws:kinesis"""
    return self["eventSource"]

The AWS service from which the Kinesis event originated. For Kinesis, this is aws:kinesis

prop event_source_arn : str

Expand source code

@property
def event_source_arn(self) -> str:
    """The Amazon Resource Name (ARN) of the event source"""
    return self["eventSourceARN"]

The Amazon Resource Name (ARN) of the event source

prop event_version : str

Expand source code

@property
def event_version(self) -> str:
    """The eventVersion key value contains a major and minor version in the form <major>.<minor>."""
    return self["eventVersion"]

The eventVersion key value contains a major and minor version in the form ..

prop invoke_identity_arn : str

Expand source code

@property
def invoke_identity_arn(self) -> str:
    """The ARN for the identity used to invoke the Lambda Function"""
    return self["invokeIdentityArn"]

The ARN for the identity used to invoke the Lambda Function

prop kinesis : KinesisStreamRecordPayload

Expand source code

@property
def kinesis(self) -> KinesisStreamRecordPayload:
    """Underlying Kinesis record associated with the event"""
    return KinesisStreamRecordPayload(self._data)

Underlying Kinesis record associated with the event

Inherited members

• DictWrapper:
  • get
  • raw_event

class KinesisStreamRecordPayload(DictWrapper):
    @property
    def approximate_arrival_timestamp(self) -> float:
        """The approximate time that the record was inserted into the stream"""
        return float(self["kinesis"]["approximateArrivalTimestamp"])

    @property
    def data(self) -> str:
        """The data blob"""
        return self["kinesis"]["data"]

    @property
    def kinesis_schema_version(self) -> str:
        """Schema version for the record"""
        return self["kinesis"]["kinesisSchemaVersion"]

    @property
    def partition_key(self) -> str:
        """Identifies which shard in the stream the data record is assigned to"""
        return self["kinesis"]["partitionKey"]

    @property
    def sequence_number(self) -> str:
        """The unique identifier of the record within its shard"""
        return self["kinesis"]["sequenceNumber"]

    def data_as_bytes(self) -> bytes:
        """Decode binary encoded data as bytes"""
        return base64.b64decode(self.data)

    def data_as_text(self) -> str:
        """Decode binary encoded data as text"""
        return self.data_as_bytes().decode("utf-8")

    def data_as_json(self) -> dict:
        """Decode binary encoded data as json"""
        return json.loads(self.data_as_text())

    def data_zlib_compressed_as_json(self) -> dict:
        """Decode binary encoded data as bytes"""
        decompressed = zlib.decompress(self.data_as_bytes(), zlib.MAX_WBITS | 32)
        return json.loads(decompressed)

Ancestors

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

Instance variables

prop approximate_arrival_timestamp : float

Expand source code

@property
def approximate_arrival_timestamp(self) -> float:
    """The approximate time that the record was inserted into the stream"""
    return float(self["kinesis"]["approximateArrivalTimestamp"])

The approximate time that the record was inserted into the stream

prop data : str

Expand source code

@property
def data(self) -> str:
    """The data blob"""
    return self["kinesis"]["data"]

The data blob

prop kinesis_schema_version : str

Expand source code

@property
def kinesis_schema_version(self) -> str:
    """Schema version for the record"""
    return self["kinesis"]["kinesisSchemaVersion"]

Schema version for the record

prop partition_key : str

Expand source code

@property
def partition_key(self) -> str:
    """Identifies which shard in the stream the data record is assigned to"""
    return self["kinesis"]["partitionKey"]

Identifies which shard in the stream the data record is assigned to

prop sequence_number : str

Expand source code

@property
def sequence_number(self) -> str:
    """The unique identifier of the record within its shard"""
    return self["kinesis"]["sequenceNumber"]

The unique identifier of the record within its shard

Methods

def data_as_bytes(self) ‑> bytes

Expand source code

def data_as_bytes(self) -> bytes:
    """Decode binary encoded data as bytes"""
    return base64.b64decode(self.data)

Decode binary encoded data as bytes

def data_as_json(self) ‑> dict

Expand source code

def data_as_json(self) -> dict:
    """Decode binary encoded data as json"""
    return json.loads(self.data_as_text())

Decode binary encoded data as json

def data_as_text(self) ‑> str

Expand source code

def data_as_text(self) -> str:
    """Decode binary encoded data as text"""
    return self.data_as_bytes().decode("utf-8")

Decode binary encoded data as text

def data_zlib_compressed_as_json(self) ‑> dict

Expand source code

def data_zlib_compressed_as_json(self) -> dict:
    """Decode binary encoded data as bytes"""
    decompressed = zlib.decompress(self.data_as_bytes(), zlib.MAX_WBITS | 32)
    return json.loads(decompressed)

Decode binary encoded data as bytes

Inherited members

• DictWrapper:
  • get
  • raw_event