22709357475

Committed 05 Mar 2026 08:35AM UTC coverage: 59.732% (-27.2%) from 86.974%

Build # 22709357475

Build Type

Pull #13880

github

Committed by

web-flow

Commit Message

Merge 28fcab93c into 710618057

Pull Request Pull Request #13880: Firehose: Replace TaggingService

Coverage Stats

12 of 12 new or added lines in 2 files covered. (100.0%)

20464 existing lines in 510 files now uncovered.

45290 of 75822 relevant lines covered (59.73%)

0.6 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

75.0

/localstack-core/localstack/services/lambda_/event_source_mapping/pollers/dynamodb_poller.py

import logging
from datetime import datetime

from botocore.client import BaseClient

from localstack.aws.api.dynamodbstreams import StreamStatus
from localstack.services.lambda_.event_source_mapping.event_processor import (
    EventProcessor,
)
from localstack.services.lambda_.event_source_mapping.pipe_utils import get_current_time
from localstack.services.lambda_.event_source_mapping.pollers.stream_poller import StreamPoller

LOG = logging.getLogger(__name__)


class DynamoDBPoller(StreamPoller):
    def __init__(
        self,
        source_arn: str,
        source_parameters: dict | None = None,
        source_client: BaseClient | None = None,
        processor: EventProcessor | None = None,
        partner_resource_arn: str | None = None,
        esm_uuid: str | None = None,
        shards: dict[str, str] | None = None,
    ):
        super().__init__(
            source_arn,
            source_parameters,
            source_client,
            processor,
            esm_uuid=esm_uuid,
            partner_resource_arn=partner_resource_arn,
            shards=shards,
        )

    @property
    def stream_parameters(self) -> dict:
        return self.source_parameters["DynamoDBStreamParameters"]

    def initialize_shards(self):
        # TODO: update upon re-sharding, maybe using a cache and call every time?!
        stream_info = self.source_client.describe_stream(StreamArn=self.source_arn)
        stream_status = stream_info["StreamDescription"]["StreamStatus"]
        if stream_status != StreamStatus.ENABLED:
            LOG.warning(
                "DynamoDB stream %s is not enabled. Current status: %s",
                self.source_arn,
                stream_status,
            )
            return {}

        # NOTICE: re-sharding might require updating this periodically (unknown how Pipes does it!?)
        # Mapping of shard id => shard iterator
        shards = {}
        for shard in stream_info["StreamDescription"]["Shards"]:
            shard_id = shard["ShardId"]
            starting_position = self.stream_parameters["StartingPosition"]
            kwargs = {}
            get_shard_iterator_response = self.source_client.get_shard_iterator(
                StreamArn=self.source_arn,
                ShardId=shard_id,
                ShardIteratorType=starting_position,
                **kwargs,
            )
            shards[shard_id] = get_shard_iterator_response["ShardIterator"]

        LOG.debug("Event source %s has %d shards.", self.source_arn, len(self.shards))
        return shards

    def stream_arn_param(self) -> dict:
        # Not supported for GetRecords:
        # https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_streams_GetRecords.html
        return {}

    def event_source(self) -> str:
        return "aws:dynamodb"

    def extra_metadata(self) -> dict:
        return {
            "eventVersion": "1.1",
        }

    def transform_into_events(self, records: list[dict], shard_id) -> list[dict]:
        events = []
        for record in records:
            # TODO: consolidate with DynamoDB event source listener:
            #  localstack.services.lambda_.event_source_listeners.dynamodb_event_source_listener.DynamoDBEventSourceListener._create_lambda_event_payload
            dynamodb = record["dynamodb"]

            if creation_time := dynamodb.get("ApproximateCreationDateTime"):
                # Float conversion validated by TestDynamoDBEventSourceMapping.test_dynamodb_event_filter
                dynamodb["ApproximateCreationDateTime"] = float(creation_time.timestamp())
            event = {
                # TODO: add this metadata after filtering (these are based on the original record!)
                #  This requires some design adjustment because the eventId and eventName depend on the record.
                "eventID": record["eventID"],
                "eventName": record["eventName"],
                # record content
                "dynamodb": dynamodb,
            }
            events.append(event)
        return events

    def failure_payload_details_field_name(self) -> str:
        return "DDBStreamBatchInfo"

    def get_approximate_arrival_time(self, record: dict) -> float:
        # TODO: validate whether the default should be now
        # Optional according to AWS docs:
        # https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_streams_StreamRecord.html
        # TODO: parse float properly if present from ApproximateCreationDateTime -> now works, compare via debug!
        return record["dynamodb"].get("todo", get_current_time().timestamp())

    def format_datetime(self, time: datetime) -> str:
        return f"{time.isoformat(timespec='seconds')}Z"

    def get_sequence_number(self, record: dict) -> str:
        return record["dynamodb"]["SequenceNumber"]

1	import logging	1✔
2	from datetime import datetime	1✔
3
4	from botocore.client import BaseClient	1✔
5
6	from localstack.aws.api.dynamodbstreams import StreamStatus	1✔
7	from localstack.services.lambda_.event_source_mapping.event_processor import (	1✔
8	EventProcessor,
9	)
10	from localstack.services.lambda_.event_source_mapping.pipe_utils import get_current_time	1✔
11	from localstack.services.lambda_.event_source_mapping.pollers.stream_poller import StreamPoller	1✔
12
13	LOG = logging.getLogger(__name__)	1✔
14
15
16	class DynamoDBPoller(StreamPoller):	1✔
17	def __init__(	1✔
18	self,
19	source_arn: str,
20	source_parameters: dict \| None = None,
21	source_client: BaseClient \| None = None,
22	processor: EventProcessor \| None = None,
23	partner_resource_arn: str \| None = None,
24	esm_uuid: str \| None = None,
25	shards: dict[str, str] \| None = None,
26	):
27	super().__init__(	1✔
28	source_arn,
29	source_parameters,
30	source_client,
31	processor,
32	esm_uuid=esm_uuid,
33	partner_resource_arn=partner_resource_arn,
34	shards=shards,
35	)
36
37	@property	1✔
38	def stream_parameters(self) -> dict:	1✔
39	return self.source_parameters["DynamoDBStreamParameters"]	1✔
40
41	def initialize_shards(self):	1✔
42	# TODO: update upon re-sharding, maybe using a cache and call every time?!
43	stream_info = self.source_client.describe_stream(StreamArn=self.source_arn)	1✔
UNCOV 44	stream_status = stream_info["StreamDescription"]["StreamStatus"]	×
UNCOV 45	if stream_status != StreamStatus.ENABLED:	×
UNCOV 46	LOG.warning(	×
47	"DynamoDB stream %s is not enabled. Current status: %s",
48	self.source_arn,
49	stream_status,
50	)
UNCOV 51	return {}	×
52
53	# NOTICE: re-sharding might require updating this periodically (unknown how Pipes does it!?)
54	# Mapping of shard id => shard iterator
UNCOV 55	shards = {}	×
56	for shard in stream_info["StreamDescription"]["Shards"]:	1✔
UNCOV 57	shard_id = shard["ShardId"]	×
UNCOV 58	starting_position = self.stream_parameters["StartingPosition"]	×
UNCOV 59	kwargs = {}	×
UNCOV 60	get_shard_iterator_response = self.source_client.get_shard_iterator(	×
61	StreamArn=self.source_arn,
62	ShardId=shard_id,
63	ShardIteratorType=starting_position,
64	**kwargs,
65	)
66	shards[shard_id] = get_shard_iterator_response["ShardIterator"]	1✔
67
68	LOG.debug("Event source %s has %d shards.", self.source_arn, len(self.shards))	1✔
69	return shards	1✔
70
71	def stream_arn_param(self) -> dict:	1✔
72	# Not supported for GetRecords:
73	# https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_streams_GetRecords.html
74	return {}	1✔
75
76	def event_source(self) -> str:	1✔
77	return "aws:dynamodb"	1✔
78
79	def extra_metadata(self) -> dict:	1✔
80	return {	1✔
81	"eventVersion": "1.1",
82	}
83
84	def transform_into_events(self, records: list[dict], shard_id) -> list[dict]:	1✔
85	events = []	1✔
86	for record in records:	1✔
87	# TODO: consolidate with DynamoDB event source listener:
88	# localstack.services.lambda_.event_source_listeners.dynamodb_event_source_listener.DynamoDBEventSourceListener._create_lambda_event_payload
89	dynamodb = record["dynamodb"]	1✔
90
91	if creation_time := dynamodb.get("ApproximateCreationDateTime"):	1✔
92	# Float conversion validated by TestDynamoDBEventSourceMapping.test_dynamodb_event_filter
93	dynamodb["ApproximateCreationDateTime"] = float(creation_time.timestamp())	1✔
94	event = {	1✔
95	# TODO: add this metadata after filtering (these are based on the original record!)
96	# This requires some design adjustment because the eventId and eventName depend on the record.
97	"eventID": record["eventID"],
98	"eventName": record["eventName"],
99	# record content
100	"dynamodb": dynamodb,
101	}
102	events.append(event)	1✔
103	return events	1✔
104
105	def failure_payload_details_field_name(self) -> str:	1✔
UNCOV 106	return "DDBStreamBatchInfo"	×
107
108	def get_approximate_arrival_time(self, record: dict) -> float:	1✔
109	# TODO: validate whether the default should be now
110	# Optional according to AWS docs:
111	# https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_streams_StreamRecord.html
112	# TODO: parse float properly if present from ApproximateCreationDateTime -> now works, compare via debug!
UNCOV 113	return record["dynamodb"].get("todo", get_current_time().timestamp())	×
114
115	def format_datetime(self, time: datetime) -> str:	1✔
UNCOV 116	return f"{time.isoformat(timespec='seconds')}Z"	×
117
118	def get_sequence_number(self, record: dict) -> str:	1✔
UNCOV 119	return record["dynamodb"]["SequenceNumber"]	×

localstack / localstack / 22709357475

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous