Much of the way towards complete DynamoDB Streams implementation

moto/__init__.py

@@ -16,6 +16,7 @@ from .cognitoidp import mock_cognitoidp, mock_cognitoidp_deprecated  # flake8: n
 from .datapipeline import mock_datapipeline, mock_datapipeline_deprecated  # flake8: noqa
 from .dynamodb import mock_dynamodb, mock_dynamodb_deprecated  # flake8: noqa
 from .dynamodb2 import mock_dynamodb2, mock_dynamodb2_deprecated  # flake8: noqa
+from .dynamodbstreams import mock_dynamodbstreams # flake8: noqa
 from .ec2 import mock_ec2, mock_ec2_deprecated  # flake8: noqa
 from .ecr import mock_ecr, mock_ecr_deprecated  # flake8: noqa
 from .ecs import mock_ecs, mock_ecs_deprecated  # flake8: noqa

moto/backends.py

@@ -12,6 +12,7 @@ from moto.core import moto_api_backends
 from moto.datapipeline import datapipeline_backends
 from moto.dynamodb import dynamodb_backends
 from moto.dynamodb2 import dynamodb_backends2
+from moto.dynamodbstreams import dynamodbstreams_backends
 from moto.ec2 import ec2_backends
 from moto.ecr import ecr_backends
 from moto.ecs import ecs_backends

moto/dynamodb2/models.py

@@ -292,6 +292,34 @@ class Item(BaseModel):
                         'ADD not supported for %s' % ', '.join(update_action['Value'].keys()))
 
 
+class StreamShard(BaseModel):
+    def __init__(self, table):
+        self.table = table
+        self.id = 'shardId-00000001541626099285-f35f62ef'
+        self.starting_sequence_number = 1100000000017454423009
+        self.items = []
+        self.created_on = datetime.datetime.utcnow()
+
+    def to_json(self):
+        return {
+            'ShardId': self.id,
+            'SequenceNumberRange': {
+                'StartingSequenceNumber': str(self.starting_sequence_number)
+            }
+        }
+
+    def add(self, old, new):
+        t = self.table.stream_specification['StreamViewType']
+        if t == 'KEYS_ONLY':
+            self.items.append(new.key)
+        elif t == 'NEW_IMAGE':
+            self.items.append(new)
+        elif t == 'OLD_IMAGE':
+            self.items.append(old)
+        elif t == 'NEW_AND_OLD_IMAGES':
+            self.items.append((old, new))
+                
+
 class Table(BaseModel):
 
     def __init__(self, table_name, schema=None, attr=None, throughput=None, indexes=None, global_indexes=None, streams=None):
@@ -334,8 +362,10 @@ class Table(BaseModel):
         self.stream_specification = streams
         if streams and streams.get('StreamEnabled'):
             self.latest_stream_label = datetime.datetime.utcnow().isoformat()
+            self.stream_shard = StreamShard(self)
         else:
             self.latest_stream_label = None
+            self.stream_shard = None
 
     def describe(self, base_key='TableDescription'):
         results = {
@@ -398,6 +428,7 @@ class Table(BaseModel):
         else:
             range_value = None
 
+        current = self.get_item(hash_value, lookup_range_value)
         item = Item(hash_value, self.hash_key_type, range_value,
                     self.range_key_type, item_attrs)
 
@@ -413,8 +444,6 @@ class Table(BaseModel):
                 else:
                     lookup_range_value = DynamoType(expected_range_value)
 
-            current = self.get_item(hash_value, lookup_range_value)
-
             if current is None:
                 current_attr = {}
             elif hasattr(current, 'attrs'):
@@ -445,6 +474,10 @@ class Table(BaseModel):
             self.items[hash_value][range_value] = item
         else:
             self.items[hash_value] = item
+            
+        if self.stream_shard is not None:
+            self.stream_shard.add(current, item)
+        
         return item
 
     def __nonzero__(self):

moto/dynamodbstreams/__init__.py

@@ -0,0 +1,6 @@
+from __future__ import unicode_literals
+from .models import dynamodbstreams_backends
+from ..core.models import base_decorator
+
+dynamodbstreams_backend = dynamodbstreams_backends['us-east-1']
+mock_dynamodbstreams = base_decorator(dynamodbstreams_backends)

moto/dynamodbstreams/models.py

@@ -0,0 +1,99 @@
+from __future__ import unicode_literals
+
+import os
+import json
+import boto3
+import base64
+import datetime
+
+from moto.core import BaseBackend, BaseModel
+from moto.dynamodb2.models import dynamodb_backends
+
+
+class ShardIterator(BaseModel):
+    def __init__(self, stream_shard, shard_iterator_type, sequence_number=None):
+        self.id = base64.b64encode(os.urandom(472)).decode('utf-8')
+        self.stream_shard = stream_shard
+        self.shard_iterator_type = shard_iterator_type
+        if shard_iterator_type == 'TRIM_HORIZON':
+            self.sequence_number = stream_shard.starting_sequence_number
+        elif shard_iterator_type == 'LATEST':
+            self.sequence_number = stream_shard.starting_sequence_number + len(stream_shard.items)
+        elif shard_iterator_type == 'AT_SEQUENCE_NUMBER':
+            self.sequence_number = sequence_number
+        elif shard_iterator_type == 'AFTER_SEQUENCE_NUMBER':
+            self.sequence_number = sequence_number + 1
+
+    def to_json(self):
+        return {
+            'ShardIterator': '{}/stream/{}|1|{}'.format(
+                self.stream_shard.table.table_arn,
+                self.stream_shard.table.latest_stream_label,
+                self.id)
+        }
+
+
+class DynamoDBStreamsBackend(BaseBackend):
+    def __init__(self, region):
+        self.region = region
+
+    def reset(self):
+        region = self.region
+        self.__dict__ = {}
+        self.__init__(region)
+        
+    @property
+    def dynamodb(self):
+        return dynamodb_backends[self.region]
+
+    def _get_table_from_arn(self, arn):
+        table_name = arn.split(':', 6)[5].split('/')[1]
+        return self.dynamodb.get_table(table_name)
+    
+    def describe_stream(self, arn):
+        table = self._get_table_from_arn(arn)
+        resp = {'StreamDescription': {
+            'StreamArn': arn,
+            'StreamLabel': table.latest_stream_label,
+            'StreamStatus': ('ENABLED' if table.latest_stream_label
+                             else 'DISABLED'),
+            'StreamViewType': table.stream_specification['StreamViewType'],
+            'CreationRequestDateTime': table.stream_shard.created_on.isoformat(),
+            'TableName': table.name,
+            'KeySchema': table.schema,
+            'Shards': ([table.stream_shard.to_json()] if table.stream_shard
+                       else [])
+        }}
+        
+        return json.dumps(resp)
+
+    def list_streams(self, table_name=None):
+        streams = []
+        for table in self.dynamodb.tables.values():
+            if table_name is not None and table.name != table_name:
+                continue
+            if table.latest_stream_label:
+                d = table.describe(base_key='Table')
+                streams.append({
+                    'StreamArn': d['Table']['LatestStreamArn'],
+                    'TableName': d['Table']['TableName'],
+                    'StreamLabel': d['Table']['LatestStreamLabel']
+                })
+                
+        return json.dumps({'Streams': streams})
+
+    def get_shard_iterator(self, arn, shard_id, shard_iterator_type, sequence_number=None):
+        table = self._get_table_from_arn(arn)
+        assert table.stream_shard.id == shard_id
+
+        shard_iterator = ShardIterator(table.stream_shard, shard_iterator_type,
+                                       sequence_number)
+        
+        return json.dumps(shard_iterator.to_json())
+
+    
+
+available_regions = boto3.session.Session().get_available_regions(
+    'dynamodbstreams')
+dynamodbstreams_backends = {region: DynamoDBStreamsBackend(region=region)
+                            for region in available_regions}

moto/dynamodbstreams/responses.py

@@ -0,0 +1,29 @@
+from __future__ import unicode_literals
+
+import json
+
+from moto.core.responses import BaseResponse
+
+from .models import dynamodbstreams_backends
+
+
+class DynamoDBStreamsHandler(BaseResponse):
+
+    @property
+    def backend(self):
+        return dynamodbstreams_backends[self.region]
+    
+    def describe_stream(self):
+        arn = self._get_param('StreamArn')
+        return self.backend.describe_stream(arn)
+
+    def list_streams(self):
+        table_name = self._get_param('TableName')
+        return self.backend.list_streams(table_name)
+
+    def get_shard_iterator(self):
+        arn = self._get_param('StreamArn')
+        shard_id = self._get_param('ShardId')
+        shard_iterator_type = self._get_param('ShardIteratorType')
+        return self.backend.get_shard_iterator(arn, shard_id,
+                                               shard_iterator_type)

moto/dynamodbstreams/urls.py

@@ -0,0 +1,10 @@
+from __future__ import unicode_literals
+from .responses import DynamoDBStreamsHandler
+
+url_bases = [
+    "https?://streams.dynamodb.(.+).amazonaws.com"
+]
+
+url_paths = {
+    "{0}/$": DynamoDBStreamsHandler.dispatch,
+}

setup.cfg

@@ -1,7 +1,7 @@
 [nosetests]
 verbosity=1
 detailed-errors=1
-with-coverage=1
+#with-coverage=1
 cover-package=moto
 
 [bdist_wheel]

tests/test_dynamodbstreams/test_dynamodbstreams.py

@@ -0,0 +1,96 @@
+from __future__ import unicode_literals, print_function
+
+import boto3
+from moto import mock_dynamodb2, mock_dynamodbstreams
+
+
+class TestClass():
+    stream_arn = None
+    mocks = []
+    
+    def setup(self):
+        self.mocks = [mock_dynamodb2(), mock_dynamodbstreams()]
+        for m in self.mocks:
+            m.start()
+            
+        # create a table with a stream
+        conn = boto3.client('dynamodb', region_name='us-east-1')
+
+        resp = conn.create_table(
+            TableName='test-streams',
+            KeySchema=[{'AttributeName': 'id', 'KeyType': 'HASH'}],
+            AttributeDefinitions=[{'AttributeName': 'id',
+                                   'AttributeType': 'S'}],
+            ProvisionedThroughput={'ReadCapacityUnits': 1,
+                                   'WriteCapacityUnits': 1},
+            StreamSpecification={
+                'StreamEnabled': True,
+                'StreamViewType': 'NEW_AND_OLD_IMAGES'
+            }
+        )
+        self.stream_arn = resp['TableDescription']['LatestStreamArn']
+
+    def teardown(self):
+        conn = boto3.client('dynamodb', region_name='us-east-1')
+        conn.delete_table(TableName='test-streams')
+        self.stream_arn = None
+
+        for m in self.mocks:
+            m.stop()
+
+
+    def test_verify_stream(self):
+        conn = boto3.client('dynamodb', region_name='us-east-1')
+        resp = conn.describe_table(TableName='test-streams')
+        assert 'LatestStreamArn' in resp['Table']
+
+    def test_describe_stream(self):
+        conn = boto3.client('dynamodbstreams', region_name='us-east-1')
+
+        resp = conn.describe_stream(StreamArn=self.stream_arn)
+        assert 'StreamDescription' in resp
+        desc = resp['StreamDescription']
+        assert desc['StreamArn'] == self.stream_arn
+        assert desc['TableName'] == 'test-streams'
+
+    def test_list_streams(self):
+        conn = boto3.client('dynamodbstreams', region_name='us-east-1')
+
+        resp = conn.list_streams()
+        assert resp['Streams'][0]['StreamArn'] == self.stream_arn
+
+        resp = conn.list_streams(TableName='no-stream')
+        assert not resp['Streams']
+
+    def test_get_shard_iterator(self):
+        conn = boto3.client('dynamodbstreams', region_name='us-east-1')
+
+        resp = conn.describe_stream(StreamArn=self.stream_arn)
+        shard_id = resp['StreamDescription']['Shards'][0]['ShardId']
+        
+        resp = conn.get_shard_iterator(
+            StreamArn=self.stream_arn,
+            ShardId=shard_id,
+            ShardIteratorType='TRIM_HORIZON'
+        )
+        assert 'ShardIterator' in resp
+                
+    def test_get_records(self):
+        conn = boto3.client('dynamodbstreams', region_name='us-east-1')
+
+        resp = conn.describe_stream(StreamArn=self.stream_arn)
+        shard_id = resp['StreamDescription']['Shards'][0]['ShardId']
+        
+        resp = conn.get_shard_iterator(
+            StreamArn=self.stream_arn,
+            ShardId=shard_id,
+            ShardIteratorType='TRIM_HORIZON'
+        )
+        iterator_id = resp['ShardIterator']
+
+        resp = conn.get_records(ShardIterator=iterator_id)
+        assert 'Records' in resp
+
+        # TODO: Add tests for inserting records into the stream, and
+        # the various stream types
+