We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
describe-stream for dynamodb streams crashes
Steps to recreate:
1. create table
```
aws dynamodb create-table --table-name MusicCollection --attribute-definitions AttributeName=Artist,AttributeType=S AttributeName=SongTitle,AttributeType=S --key-schema AttributeName=Artist,KeyType=HASH AttributeName=SongTitle,KeyType=RANGE --provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5 --endpoint http://localhost:4569
```
2. add stream
```
aws dynamodb update-table --table-name MusicCollection --stream-specification StreamEnabled=true,StreamViewType=NEW_IMAGE --endpoint http://localhost:4569  
```
3. describe stream using the stream arn from step 2.
```
aws dynamodbstreams describe-stream --stream-arn "arn:aws:dynamodb:eu-central-1:000000000000:table/MusicCollection/stream/2019-09-21T14:06:37.345" --endpoint http://localhost:4570
```
localstack logs:

```
Traceback (most recent call last):
  File "/home/balrog/localstack/.venv/lib/python2.7/site-packages/flask/app.py", line 2292, in wsgi_app
    response = self.full_dispatch_request()
  File "/home/balrog/localstack/.venv/lib/python2.7/site-packages/flask/app.py", line 1815, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/home/balrog/localstack/.venv/lib/python2.7/site-packages/flask_cors/extension.py", line 161, in wrapped_function
    return cors_after_request(app.make_response(f(*args, **kwargs)))
  File "/home/balrog/localstack/.venv/lib/python2.7/site-packages/flask/app.py", line 1718, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/home/balrog/localstack/.venv/lib/python2.7/site-packages/flask/app.py", line 1813, in full_dispatch_request
    rv = self.dispatch_request()
  File "/home/balrog/localstack/.venv/lib/python2.7/site-packages/flask/app.py", line 1799, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/home/balrog/localstack/localstack/services/dynamodbstreams/dynamodbstreams_api.py", line 88, in post_request
    shard['ShardId'] = shard_id(stream_name, shard['ShardId'])
  File "/home/balrog/localstack/localstack/services/dynamodbstreams/dynamodbstreams_api.py", line 146, in shard_id
    return '-'.join([kinesis_shard_id, random_id(stream_arn, kinesis_shard_id)])
  File "/home/balrog/localstack/localstack/services/dynamodbstreams/dynamodbstreams_api.py", line 142, in random_id
    return uuid.uuid5(namespace, kinesis_shard_id).hex
  File "/usr/lib/python2.7/uuid.py", line 609, in uuid5
    hash = sha1(namespace.bytes + name).digest()
UnicodeDecodeError: 'ascii' codec can't decode byte 0xe1 in position 4: ordinal not in range(128)
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `localstack/services/dynamodbstreams/dynamodbstreams_api.py`
Content:
```
1 import json
2 import uuid
3 import hashlib
4 from flask import Flask, jsonify, request, make_response
5 from localstack.services import generic_proxy
6 from localstack.utils.aws import aws_stack
7 from localstack.utils.common import to_str, to_bytes
8 from localstack.utils.analytics import event_publisher
9 
10 APP_NAME = 'ddb_streams_api'
11 
12 app = Flask(APP_NAME)
13 
14 DDB_STREAMS = {}
15 
16 DDB_KINESIS_STREAM_NAME_PREFIX = '__ddb_stream_'
17 
18 ACTION_HEADER_PREFIX = 'DynamoDBStreams_20120810'
19 
20 SEQUENCE_NUMBER_COUNTER = 1
21 
22 
23 def add_dynamodb_stream(table_name, latest_stream_label, view_type='NEW_AND_OLD_IMAGES', enabled=True):
24     if enabled:
25         # create kinesis stream as a backend
26         stream_name = get_kinesis_stream_name(table_name)
27         aws_stack.create_kinesis_stream(stream_name)
28         stream = {
29             'StreamArn': aws_stack.dynamodb_stream_arn(table_name=table_name,
30                                                        latest_stream_label=latest_stream_label),
31             'TableName': table_name,
32             'StreamLabel': latest_stream_label,
33             'StreamStatus': 'ENABLED',
34             'KeySchema': [],
35             'Shards': []
36         }
37         table_arn = aws_stack.dynamodb_table_arn(table_name)
38         DDB_STREAMS[table_arn] = stream
39         # record event
40         event_publisher.fire_event(event_publisher.EVENT_DYNAMODB_CREATE_STREAM,
41             payload={'n': event_publisher.get_hash(table_name)})
42 
43 
44 def forward_events(records):
45     global SEQUENCE_NUMBER_COUNTER
46     kinesis = aws_stack.connect_to_service('kinesis')
47     for record in records:
48         if 'SequenceNumber' not in record['dynamodb']:
49             record['dynamodb']['SequenceNumber'] = str(SEQUENCE_NUMBER_COUNTER)
50             SEQUENCE_NUMBER_COUNTER += 1
51         table_arn = record['eventSourceARN']
52         stream = DDB_STREAMS.get(table_arn)
53         if stream:
54             table_name = table_name_from_stream_arn(stream['StreamArn'])
55             stream_name = get_kinesis_stream_name(table_name)
56             kinesis.put_record(StreamName=stream_name, Data=json.dumps(record), PartitionKey='TODO')
57 
58 
59 @app.route('/', methods=['POST'])
60 def post_request():
61     action = request.headers.get('x-amz-target')
62     data = json.loads(to_str(request.data))
63     result = {}
64     kinesis = aws_stack.connect_to_service('kinesis')
65     if action == '%s.ListStreams' % ACTION_HEADER_PREFIX:
66         result = {
67             'Streams': list(DDB_STREAMS.values()),
68             'LastEvaluatedStreamArn': 'TODO'
69         }
70     elif action == '%s.DescribeStream' % ACTION_HEADER_PREFIX:
71         for stream in DDB_STREAMS.values():
72             if stream['StreamArn'] == data['StreamArn']:
73                 result = {
74                     'StreamDescription': stream
75                 }
76                 # get stream details
77                 dynamodb = aws_stack.connect_to_service('dynamodb')
78                 table_name = table_name_from_stream_arn(stream['StreamArn'])
79                 stream_name = get_kinesis_stream_name(table_name)
80                 stream_details = kinesis.describe_stream(StreamName=stream_name)
81                 table_details = dynamodb.describe_table(TableName=table_name)
82                 stream['KeySchema'] = table_details['Table']['KeySchema']
83 
84                 # Replace Kinesis ShardIDs with ones that mimic actual
85                 # DynamoDBStream ShardIDs.
86                 stream_shards = stream_details['StreamDescription']['Shards']
87                 for shard in stream_shards:
88                     shard['ShardId'] = shard_id(stream_name, shard['ShardId'])
89                 stream['Shards'] = stream_shards
90                 break
91         if not result:
92             return error_response('Requested resource not found', error_type='ResourceNotFoundException')
93     elif action == '%s.GetShardIterator' % ACTION_HEADER_PREFIX:
94         # forward request to Kinesis API
95         stream_name = stream_name_from_stream_arn(data['StreamArn'])
96         stream_shard_id = kinesis_shard_id(data['ShardId'])
97         result = kinesis.get_shard_iterator(StreamName=stream_name,
98             ShardId=stream_shard_id, ShardIteratorType=data['ShardIteratorType'])
99     elif action == '%s.GetRecords' % ACTION_HEADER_PREFIX:
100         kinesis_records = kinesis.get_records(**data)
101         result = {'Records': [], 'NextShardIterator': kinesis_records.get('NextShardIterator')}
102         for record in kinesis_records['Records']:
103             result['Records'].append(json.loads(to_str(record['Data'])))
104     else:
105         print('WARNING: Unknown operation "%s"' % action)
106     return jsonify(result)
107 
108 
109 # -----------------
110 # HELPER FUNCTIONS
111 # -----------------
112 
113 def error_response(message=None, error_type=None, code=400):
114     if not message:
115         message = 'Unknown error'
116     if not error_type:
117         error_type = 'UnknownError'
118     if 'com.amazonaws.dynamodb' not in error_type:
119         error_type = 'com.amazonaws.dynamodb.v20120810#%s' % error_type
120     content = {
121         'message': message,
122         '__type': error_type
123     }
124     return make_response(jsonify(content), code)
125 
126 
127 def get_kinesis_stream_name(table_name):
128     return DDB_KINESIS_STREAM_NAME_PREFIX + table_name
129 
130 
131 def table_name_from_stream_arn(stream_arn):
132     return stream_arn.split(':table/')[1].split('/')[0]
133 
134 
135 def stream_name_from_stream_arn(stream_arn):
136     table_name = table_name_from_stream_arn(stream_arn)
137     return get_kinesis_stream_name(table_name)
138 
139 
140 def random_id(stream_arn, kinesis_shard_id):
141     namespace = uuid.UUID(bytes=hashlib.sha1(to_bytes(stream_arn)).digest()[:16])
142     return uuid.uuid5(namespace, kinesis_shard_id).hex
143 
144 
145 def shard_id(stream_arn, kinesis_shard_id):
146     return '-'.join([kinesis_shard_id, random_id(stream_arn, kinesis_shard_id)])
147 
148 
149 def kinesis_shard_id(dynamodbstream_shard_id):
150     return dynamodbstream_shard_id.rsplit('-', 1)[0]
151 
152 
153 def serve(port, quiet=True):
154     generic_proxy.serve_flask_app(app=app, port=port, quiet=quiet)
155
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/localstack/services/dynamodbstreams/dynamodbstreams_api.py b/localstack/services/dynamodbstreams/dynamodbstreams_api.py
--- a/localstack/services/dynamodbstreams/dynamodbstreams_api.py
+++ b/localstack/services/dynamodbstreams/dynamodbstreams_api.py
@@ -139,7 +139,7 @@
 
 def random_id(stream_arn, kinesis_shard_id):
     namespace = uuid.UUID(bytes=hashlib.sha1(to_bytes(stream_arn)).digest()[:16])
-    return uuid.uuid5(namespace, kinesis_shard_id).hex
+    return uuid.uuid5(namespace, to_bytes(kinesis_shard_id)).hex
 
 
 def shard_id(stream_arn, kinesis_shard_id):

{"golden_diff": "diff --git a/localstack/services/dynamodbstreams/dynamodbstreams_api.py b/localstack/services/dynamodbstreams/dynamodbstreams_api.py\n--- a/localstack/services/dynamodbstreams/dynamodbstreams_api.py\n+++ b/localstack/services/dynamodbstreams/dynamodbstreams_api.py\n@@ -139,7 +139,7 @@\n \n def random_id(stream_arn, kinesis_shard_id):\n     namespace = uuid.UUID(bytes=hashlib.sha1(to_bytes(stream_arn)).digest()[:16])\n-    return uuid.uuid5(namespace, kinesis_shard_id).hex\n+    return uuid.uuid5(namespace, to_bytes(kinesis_shard_id)).hex\n \n \n def shard_id(stream_arn, kinesis_shard_id):\n", "issue": "describe-stream for dynamodb streams crashes\nSteps to recreate:\r\n1. create table\r\n```\r\naws dynamodb create-table --table-name MusicCollection --attribute-definitions AttributeName=Artist,AttributeType=S AttributeName=SongTitle,AttributeType=S --key-schema AttributeName=Artist,KeyType=HASH AttributeName=SongTitle,KeyType=RANGE --provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5 --endpoint http://localhost:4569\r\n```\r\n2. add stream\r\n```\r\naws dynamodb update-table --table-name MusicCollection --stream-specification StreamEnabled=true,StreamViewType=NEW_IMAGE --endpoint http://localhost:4569  \r\n```\r\n3. describe stream using the stream arn from step 2.\r\n```\r\naws dynamodbstreams describe-stream --stream-arn \"arn:aws:dynamodb:eu-central-1:000000000000:table/MusicCollection/stream/2019-09-21T14:06:37.345\" --endpoint http://localhost:4570\r\n```\r\nlocalstack logs:\r\n\r\n```\r\nTraceback (most recent call last):\r\n  File \"/home/balrog/localstack/.venv/lib/python2.7/site-packages/flask/app.py\", line 2292, in wsgi_app\r\n    response = self.full_dispatch_request()\r\n  File \"/home/balrog/localstack/.venv/lib/python2.7/site-packages/flask/app.py\", line 1815, in full_dispatch_request\r\n    rv = self.handle_user_exception(e)\r\n  File \"/home/balrog/localstack/.venv/lib/python2.7/site-packages/flask_cors/extension.py\", line 161, in wrapped_function\r\n    return cors_after_request(app.make_response(f(*args, **kwargs)))\r\n  File \"/home/balrog/localstack/.venv/lib/python2.7/site-packages/flask/app.py\", line 1718, in handle_user_exception\r\n    reraise(exc_type, exc_value, tb)\r\n  File \"/home/balrog/localstack/.venv/lib/python2.7/site-packages/flask/app.py\", line 1813, in full_dispatch_request\r\n    rv = self.dispatch_request()\r\n  File \"/home/balrog/localstack/.venv/lib/python2.7/site-packages/flask/app.py\", line 1799, in dispatch_request\r\n    return self.view_functions[rule.endpoint](**req.view_args)\r\n  File \"/home/balrog/localstack/localstack/services/dynamodbstreams/dynamodbstreams_api.py\", line 88, in post_request\r\n    shard['ShardId'] = shard_id(stream_name, shard['ShardId'])\r\n  File \"/home/balrog/localstack/localstack/services/dynamodbstreams/dynamodbstreams_api.py\", line 146, in shard_id\r\n    return '-'.join([kinesis_shard_id, random_id(stream_arn, kinesis_shard_id)])\r\n  File \"/home/balrog/localstack/localstack/services/dynamodbstreams/dynamodbstreams_api.py\", line 142, in random_id\r\n    return uuid.uuid5(namespace, kinesis_shard_id).hex\r\n  File \"/usr/lib/python2.7/uuid.py\", line 609, in uuid5\r\n    hash = sha1(namespace.bytes + name).digest()\r\nUnicodeDecodeError: 'ascii' codec can't decode byte 0xe1 in position 4: ordinal not in range(128)\r\n```\n", "before_files": [{"content": "import json\nimport uuid\nimport hashlib\nfrom flask import Flask, jsonify, request, make_response\nfrom localstack.services import generic_proxy\nfrom localstack.utils.aws import aws_stack\nfrom localstack.utils.common import to_str, to_bytes\nfrom localstack.utils.analytics import event_publisher\n\nAPP_NAME = 'ddb_streams_api'\n\napp = Flask(APP_NAME)\n\nDDB_STREAMS = {}\n\nDDB_KINESIS_STREAM_NAME_PREFIX = '__ddb_stream_'\n\nACTION_HEADER_PREFIX = 'DynamoDBStreams_20120810'\n\nSEQUENCE_NUMBER_COUNTER = 1\n\n\ndef add_dynamodb_stream(table_name, latest_stream_label, view_type='NEW_AND_OLD_IMAGES', enabled=True):\n    if enabled:\n        # create kinesis stream as a backend\n        stream_name = get_kinesis_stream_name(table_name)\n        aws_stack.create_kinesis_stream(stream_name)\n        stream = {\n            'StreamArn': aws_stack.dynamodb_stream_arn(table_name=table_name,\n                                                       latest_stream_label=latest_stream_label),\n            'TableName': table_name,\n            'StreamLabel': latest_stream_label,\n            'StreamStatus': 'ENABLED',\n            'KeySchema': [],\n            'Shards': []\n        }\n        table_arn = aws_stack.dynamodb_table_arn(table_name)\n        DDB_STREAMS[table_arn] = stream\n        # record event\n        event_publisher.fire_event(event_publisher.EVENT_DYNAMODB_CREATE_STREAM,\n            payload={'n': event_publisher.get_hash(table_name)})\n\n\ndef forward_events(records):\n    global SEQUENCE_NUMBER_COUNTER\n    kinesis = aws_stack.connect_to_service('kinesis')\n    for record in records:\n        if 'SequenceNumber' not in record['dynamodb']:\n            record['dynamodb']['SequenceNumber'] = str(SEQUENCE_NUMBER_COUNTER)\n            SEQUENCE_NUMBER_COUNTER += 1\n        table_arn = record['eventSourceARN']\n        stream = DDB_STREAMS.get(table_arn)\n        if stream:\n            table_name = table_name_from_stream_arn(stream['StreamArn'])\n            stream_name = get_kinesis_stream_name(table_name)\n            kinesis.put_record(StreamName=stream_name, Data=json.dumps(record), PartitionKey='TODO')\n\n\n@app.route('/', methods=['POST'])\ndef post_request():\n    action = request.headers.get('x-amz-target')\n    data = json.loads(to_str(request.data))\n    result = {}\n    kinesis = aws_stack.connect_to_service('kinesis')\n    if action == '%s.ListStreams' % ACTION_HEADER_PREFIX:\n        result = {\n            'Streams': list(DDB_STREAMS.values()),\n            'LastEvaluatedStreamArn': 'TODO'\n        }\n    elif action == '%s.DescribeStream' % ACTION_HEADER_PREFIX:\n        for stream in DDB_STREAMS.values():\n            if stream['StreamArn'] == data['StreamArn']:\n                result = {\n                    'StreamDescription': stream\n                }\n                # get stream details\n                dynamodb = aws_stack.connect_to_service('dynamodb')\n                table_name = table_name_from_stream_arn(stream['StreamArn'])\n                stream_name = get_kinesis_stream_name(table_name)\n                stream_details = kinesis.describe_stream(StreamName=stream_name)\n                table_details = dynamodb.describe_table(TableName=table_name)\n                stream['KeySchema'] = table_details['Table']['KeySchema']\n\n                # Replace Kinesis ShardIDs with ones that mimic actual\n                # DynamoDBStream ShardIDs.\n                stream_shards = stream_details['StreamDescription']['Shards']\n                for shard in stream_shards:\n                    shard['ShardId'] = shard_id(stream_name, shard['ShardId'])\n                stream['Shards'] = stream_shards\n                break\n        if not result:\n            return error_response('Requested resource not found', error_type='ResourceNotFoundException')\n    elif action == '%s.GetShardIterator' % ACTION_HEADER_PREFIX:\n        # forward request to Kinesis API\n        stream_name = stream_name_from_stream_arn(data['StreamArn'])\n        stream_shard_id = kinesis_shard_id(data['ShardId'])\n        result = kinesis.get_shard_iterator(StreamName=stream_name,\n            ShardId=stream_shard_id, ShardIteratorType=data['ShardIteratorType'])\n    elif action == '%s.GetRecords' % ACTION_HEADER_PREFIX:\n        kinesis_records = kinesis.get_records(**data)\n        result = {'Records': [], 'NextShardIterator': kinesis_records.get('NextShardIterator')}\n        for record in kinesis_records['Records']:\n            result['Records'].append(json.loads(to_str(record['Data'])))\n    else:\n        print('WARNING: Unknown operation \"%s\"' % action)\n    return jsonify(result)\n\n\n# -----------------\n# HELPER FUNCTIONS\n# -----------------\n\ndef error_response(message=None, error_type=None, code=400):\n    if not message:\n        message = 'Unknown error'\n    if not error_type:\n        error_type = 'UnknownError'\n    if 'com.amazonaws.dynamodb' not in error_type:\n        error_type = 'com.amazonaws.dynamodb.v20120810#%s' % error_type\n    content = {\n        'message': message,\n        '__type': error_type\n    }\n    return make_response(jsonify(content), code)\n\n\ndef get_kinesis_stream_name(table_name):\n    return DDB_KINESIS_STREAM_NAME_PREFIX + table_name\n\n\ndef table_name_from_stream_arn(stream_arn):\n    return stream_arn.split(':table/')[1].split('/')[0]\n\n\ndef stream_name_from_stream_arn(stream_arn):\n    table_name = table_name_from_stream_arn(stream_arn)\n    return get_kinesis_stream_name(table_name)\n\n\ndef random_id(stream_arn, kinesis_shard_id):\n    namespace = uuid.UUID(bytes=hashlib.sha1(to_bytes(stream_arn)).digest()[:16])\n    return uuid.uuid5(namespace, kinesis_shard_id).hex\n\n\ndef shard_id(stream_arn, kinesis_shard_id):\n    return '-'.join([kinesis_shard_id, random_id(stream_arn, kinesis_shard_id)])\n\n\ndef kinesis_shard_id(dynamodbstream_shard_id):\n    return dynamodbstream_shard_id.rsplit('-', 1)[0]\n\n\ndef serve(port, quiet=True):\n    generic_proxy.serve_flask_app(app=app, port=port, quiet=quiet)\n", "path": "localstack/services/dynamodbstreams/dynamodbstreams_api.py"}], "after_files": [{"content": "import json\nimport uuid\nimport hashlib\nfrom flask import Flask, jsonify, request, make_response\nfrom localstack.services import generic_proxy\nfrom localstack.utils.aws import aws_stack\nfrom localstack.utils.common import to_str, to_bytes\nfrom localstack.utils.analytics import event_publisher\n\nAPP_NAME = 'ddb_streams_api'\n\napp = Flask(APP_NAME)\n\nDDB_STREAMS = {}\n\nDDB_KINESIS_STREAM_NAME_PREFIX = '__ddb_stream_'\n\nACTION_HEADER_PREFIX = 'DynamoDBStreams_20120810'\n\nSEQUENCE_NUMBER_COUNTER = 1\n\n\ndef add_dynamodb_stream(table_name, latest_stream_label, view_type='NEW_AND_OLD_IMAGES', enabled=True):\n    if enabled:\n        # create kinesis stream as a backend\n        stream_name = get_kinesis_stream_name(table_name)\n        aws_stack.create_kinesis_stream(stream_name)\n        stream = {\n            'StreamArn': aws_stack.dynamodb_stream_arn(table_name=table_name,\n                                                       latest_stream_label=latest_stream_label),\n            'TableName': table_name,\n            'StreamLabel': latest_stream_label,\n            'StreamStatus': 'ENABLED',\n            'KeySchema': [],\n            'Shards': []\n        }\n        table_arn = aws_stack.dynamodb_table_arn(table_name)\n        DDB_STREAMS[table_arn] = stream\n        # record event\n        event_publisher.fire_event(event_publisher.EVENT_DYNAMODB_CREATE_STREAM,\n            payload={'n': event_publisher.get_hash(table_name)})\n\n\ndef forward_events(records):\n    global SEQUENCE_NUMBER_COUNTER\n    kinesis = aws_stack.connect_to_service('kinesis')\n    for record in records:\n        if 'SequenceNumber' not in record['dynamodb']:\n            record['dynamodb']['SequenceNumber'] = str(SEQUENCE_NUMBER_COUNTER)\n            SEQUENCE_NUMBER_COUNTER += 1\n        table_arn = record['eventSourceARN']\n        stream = DDB_STREAMS.get(table_arn)\n        if stream:\n            table_name = table_name_from_stream_arn(stream['StreamArn'])\n            stream_name = get_kinesis_stream_name(table_name)\n            kinesis.put_record(StreamName=stream_name, Data=json.dumps(record), PartitionKey='TODO')\n\n\n@app.route('/', methods=['POST'])\ndef post_request():\n    action = request.headers.get('x-amz-target')\n    data = json.loads(to_str(request.data))\n    result = {}\n    kinesis = aws_stack.connect_to_service('kinesis')\n    if action == '%s.ListStreams' % ACTION_HEADER_PREFIX:\n        result = {\n            'Streams': list(DDB_STREAMS.values()),\n            'LastEvaluatedStreamArn': 'TODO'\n        }\n    elif action == '%s.DescribeStream' % ACTION_HEADER_PREFIX:\n        for stream in DDB_STREAMS.values():\n            if stream['StreamArn'] == data['StreamArn']:\n                result = {\n                    'StreamDescription': stream\n                }\n                # get stream details\n                dynamodb = aws_stack.connect_to_service('dynamodb')\n                table_name = table_name_from_stream_arn(stream['StreamArn'])\n                stream_name = get_kinesis_stream_name(table_name)\n                stream_details = kinesis.describe_stream(StreamName=stream_name)\n                table_details = dynamodb.describe_table(TableName=table_name)\n                stream['KeySchema'] = table_details['Table']['KeySchema']\n\n                # Replace Kinesis ShardIDs with ones that mimic actual\n                # DynamoDBStream ShardIDs.\n                stream_shards = stream_details['StreamDescription']['Shards']\n                for shard in stream_shards:\n                    shard['ShardId'] = shard_id(stream_name, shard['ShardId'])\n                stream['Shards'] = stream_shards\n                break\n        if not result:\n            return error_response('Requested resource not found', error_type='ResourceNotFoundException')\n    elif action == '%s.GetShardIterator' % ACTION_HEADER_PREFIX:\n        # forward request to Kinesis API\n        stream_name = stream_name_from_stream_arn(data['StreamArn'])\n        stream_shard_id = kinesis_shard_id(data['ShardId'])\n        result = kinesis.get_shard_iterator(StreamName=stream_name,\n            ShardId=stream_shard_id, ShardIteratorType=data['ShardIteratorType'])\n    elif action == '%s.GetRecords' % ACTION_HEADER_PREFIX:\n        kinesis_records = kinesis.get_records(**data)\n        result = {'Records': [], 'NextShardIterator': kinesis_records.get('NextShardIterator')}\n        for record in kinesis_records['Records']:\n            result['Records'].append(json.loads(to_str(record['Data'])))\n    else:\n        print('WARNING: Unknown operation \"%s\"' % action)\n    return jsonify(result)\n\n\n# -----------------\n# HELPER FUNCTIONS\n# -----------------\n\ndef error_response(message=None, error_type=None, code=400):\n    if not message:\n        message = 'Unknown error'\n    if not error_type:\n        error_type = 'UnknownError'\n    if 'com.amazonaws.dynamodb' not in error_type:\n        error_type = 'com.amazonaws.dynamodb.v20120810#%s' % error_type\n    content = {\n        'message': message,\n        '__type': error_type\n    }\n    return make_response(jsonify(content), code)\n\n\ndef get_kinesis_stream_name(table_name):\n    return DDB_KINESIS_STREAM_NAME_PREFIX + table_name\n\n\ndef table_name_from_stream_arn(stream_arn):\n    return stream_arn.split(':table/')[1].split('/')[0]\n\n\ndef stream_name_from_stream_arn(stream_arn):\n    table_name = table_name_from_stream_arn(stream_arn)\n    return get_kinesis_stream_name(table_name)\n\n\ndef random_id(stream_arn, kinesis_shard_id):\n    namespace = uuid.UUID(bytes=hashlib.sha1(to_bytes(stream_arn)).digest()[:16])\n    return uuid.uuid5(namespace, to_bytes(kinesis_shard_id)).hex\n\n\ndef shard_id(stream_arn, kinesis_shard_id):\n    return '-'.join([kinesis_shard_id, random_id(stream_arn, kinesis_shard_id)])\n\n\ndef kinesis_shard_id(dynamodbstream_shard_id):\n    return dynamodbstream_shard_id.rsplit('-', 1)[0]\n\n\ndef serve(port, quiet=True):\n    generic_proxy.serve_flask_app(app=app, port=port, quiet=quiet)\n", "path": "localstack/services/dynamodbstreams/dynamodbstreams_api.py"}]}