shard_iterators.py

    def _get_shard_iterator(self, shard, stream_arn):
        return self.client.get_shard_iterator(
            StreamArn=stream_arn,
            ShardId=shard["ShardId"],
            ShardIteratorType="TRIM_HORIZON",
        )

    def _process_signals(self, shards, remaining_items, termination_moment):
        for shard_iterator in shards:
            while remaining_items and dt.datetime.utcnow() <= termination_moment:
                records_response = self.client.get_records(ShardIterator=shard_iterator)
                if records_response["Records"]:
                    records = records_response["Records"]
                    for j in records:
                        key = dynamo_formatter.loads(j["dynamodb"]["Keys"])
                        key = key["Id"]
                        if key in remaining_items:                          
                            remaining_items.remove(key)              

                shard_iterator = records_response.get("NextShardIterator")
                if not shard_iterator:
                    break

                time.sleep(1)
        return remaining_items