We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
describe-stream crashes in docker-image
steps to reproduce:
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
```
Enable stream:
```
aws dynamodb update-table --table-name MusicCollection --stream-specification StreamEnabled=true,StreamViewType=NEW_IMAGE --endpoint http://localhost:4569
```
describe-stream using the stream arn from previous step:
```
aws dynamodbstreams describe-stream --stream-arn arn:aws:dynamodb:eu-central-1:000000000000:table/MusicCollection/stream/2019-09-29T12:25:50.316 --endpoint http://localhost:4570
```
crash:
```
Traceback (most recent call last):
  File "/opt/code/localstack/.venv/lib/python3.6/site-packages/flask/app.py", line 2292, in wsgi_app
    response = self.full_dispatch_request()
  File "/opt/code/localstack/.venv/lib/python3.6/site-packages/flask/app.py", line 1815, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/opt/code/localstack/.venv/lib/python3.6/site-packages/flask_cors/extension.py", line 161, in wrapped_function
    return cors_after_request(app.make_response(f(*args, **kwargs)))
  File "/opt/code/localstack/.venv/lib/python3.6/site-packages/flask/app.py", line 1718, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/opt/code/localstack/.venv/lib/python3.6/site-packages/flask/_compat.py", line 35, in reraise
    raise value
  File "/opt/code/localstack/.venv/lib/python3.6/site-packages/flask/app.py", line 1813, in full_dispatch_request
    rv = self.dispatch_request()
  File "/opt/code/localstack/.venv/lib/python3.6/site-packages/flask/app.py", line 1799, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/opt/code/localstack/localstack/services/dynamodbstreams/dynamodbstreams_api.py", line 89, in post_request
    shard['ShardId'] = shard_id(stream_name, shard['ShardId'])
  File "/opt/code/localstack/localstack/services/dynamodbstreams/dynamodbstreams_api.py", line 148, in shard_id
    return '-'.join([kinesis_shard_id, random_id(stream_arn, kinesis_shard_id)])
  File "/opt/code/localstack/localstack/services/dynamodbstreams/dynamodbstreams_api.py", line 144, in random_id
    return uuid.uuid5(namespace, to_bytes(kinesis_shard_id)).hex
  File "/usr/lib/python3.6/uuid.py", line 628, in uuid5
    hash = sha1(namespace.bytes + bytes(name, "utf-8")).digest()
TypeError: encoding without a string argument
```

--- 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=None, 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         latest_stream_label = latest_stream_label or 'latest'
29         stream = {
30             'StreamArn': aws_stack.dynamodb_stream_arn(
31                 table_name=table_name, latest_stream_label=latest_stream_label),
32             'TableName': table_name,
33             'StreamLabel': latest_stream_label,
34             'StreamStatus': 'ENABLED',
35             'KeySchema': [],
36             'Shards': []
37         }
38         table_arn = aws_stack.dynamodb_table_arn(table_name)
39         DDB_STREAMS[table_arn] = stream
40         # record event
41         event_publisher.fire_event(event_publisher.EVENT_DYNAMODB_CREATE_STREAM,
42             payload={'n': event_publisher.get_hash(table_name)})
43 
44 
45 def forward_events(records):
46     global SEQUENCE_NUMBER_COUNTER
47     kinesis = aws_stack.connect_to_service('kinesis')
48     for record in records:
49         if 'SequenceNumber' not in record['dynamodb']:
50             record['dynamodb']['SequenceNumber'] = str(SEQUENCE_NUMBER_COUNTER)
51             SEQUENCE_NUMBER_COUNTER += 1
52         table_arn = record['eventSourceARN']
53         stream = DDB_STREAMS.get(table_arn)
54         if stream:
55             table_name = table_name_from_stream_arn(stream['StreamArn'])
56             stream_name = get_kinesis_stream_name(table_name)
57             kinesis.put_record(StreamName=stream_name, Data=json.dumps(record), PartitionKey='TODO')
58 
59 
60 @app.route('/', methods=['POST'])
61 def post_request():
62     action = request.headers.get('x-amz-target')
63     data = json.loads(to_str(request.data))
64     result = {}
65     kinesis = aws_stack.connect_to_service('kinesis')
66     if action == '%s.ListStreams' % ACTION_HEADER_PREFIX:
67         result = {
68             'Streams': list(DDB_STREAMS.values()),
69             'LastEvaluatedStreamArn': 'TODO'
70         }
71     elif action == '%s.DescribeStream' % ACTION_HEADER_PREFIX:
72         for stream in DDB_STREAMS.values():
73             if stream['StreamArn'] == data['StreamArn']:
74                 result = {
75                     'StreamDescription': stream
76                 }
77                 # get stream details
78                 dynamodb = aws_stack.connect_to_service('dynamodb')
79                 table_name = table_name_from_stream_arn(stream['StreamArn'])
80                 stream_name = get_kinesis_stream_name(table_name)
81                 stream_details = kinesis.describe_stream(StreamName=stream_name)
82                 table_details = dynamodb.describe_table(TableName=table_name)
83                 stream['KeySchema'] = table_details['Table']['KeySchema']
84 
85                 # Replace Kinesis ShardIDs with ones that mimic actual
86                 # DynamoDBStream ShardIDs.
87                 stream_shards = stream_details['StreamDescription']['Shards']
88                 for shard in stream_shards:
89                     shard['ShardId'] = shard_id(stream_name, shard['ShardId'])
90                 stream['Shards'] = stream_shards
91                 break
92         if not result:
93             return error_response('Requested resource not found', error_type='ResourceNotFoundException')
94     elif action == '%s.GetShardIterator' % ACTION_HEADER_PREFIX:
95         # forward request to Kinesis API
96         stream_name = stream_name_from_stream_arn(data['StreamArn'])
97         stream_shard_id = kinesis_shard_id(data['ShardId'])
98         result = kinesis.get_shard_iterator(StreamName=stream_name,
99             ShardId=stream_shard_id, ShardIteratorType=data['ShardIteratorType'])
100     elif action == '%s.GetRecords' % ACTION_HEADER_PREFIX:
101         kinesis_records = kinesis.get_records(**data)
102         result = {'Records': [], 'NextShardIterator': kinesis_records.get('NextShardIterator')}
103         for record in kinesis_records['Records']:
104             result['Records'].append(json.loads(to_str(record['Data'])))
105     else:
106         print('WARNING: Unknown operation "%s"' % action)
107     return jsonify(result)
108 
109 
110 # -----------------
111 # HELPER FUNCTIONS
112 # -----------------
113 
114 def error_response(message=None, error_type=None, code=400):
115     if not message:
116         message = 'Unknown error'
117     if not error_type:
118         error_type = 'UnknownError'
119     if 'com.amazonaws.dynamodb' not in error_type:
120         error_type = 'com.amazonaws.dynamodb.v20120810#%s' % error_type
121     content = {
122         'message': message,
123         '__type': error_type
124     }
125     return make_response(jsonify(content), code)
126 
127 
128 def get_kinesis_stream_name(table_name):
129     return DDB_KINESIS_STREAM_NAME_PREFIX + table_name
130 
131 
132 def table_name_from_stream_arn(stream_arn):
133     return stream_arn.split(':table/')[1].split('/')[0]
134 
135 
136 def stream_name_from_stream_arn(stream_arn):
137     table_name = table_name_from_stream_arn(stream_arn)
138     return get_kinesis_stream_name(table_name)
139 
140 
141 def random_id(stream_arn, kinesis_shard_id):
142     namespace = uuid.UUID(bytes=hashlib.sha1(to_bytes(stream_arn)).digest()[:16])
143     return uuid.uuid5(namespace, to_bytes(kinesis_shard_id)).hex
144 
145 
146 def shard_id(stream_arn, kinesis_shard_id):
147     return '-'.join([kinesis_shard_id, random_id(stream_arn, kinesis_shard_id)])
148 
149 
150 def kinesis_shard_id(dynamodbstream_shard_id):
151     return dynamodbstream_shard_id.rsplit('-', 1)[0]
152 
153 
154 def serve(port, quiet=True):
155     generic_proxy.serve_flask_app(app=app, port=port, quiet=quiet)
156
```
--- 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
@@ -1,6 +1,7 @@
 import json
 import uuid
 import hashlib
+import six
 from flask import Flask, jsonify, request, make_response
 from localstack.services import generic_proxy
 from localstack.utils.aws import aws_stack
@@ -140,7 +141,9 @@
 
 def random_id(stream_arn, kinesis_shard_id):
     namespace = uuid.UUID(bytes=hashlib.sha1(to_bytes(stream_arn)).digest()[:16])
-    return uuid.uuid5(namespace, to_bytes(kinesis_shard_id)).hex
+    if six.PY2:
+        kinesis_shard_id = to_bytes(kinesis_shard_id, 'utf-8')
+    return uuid.uuid5(namespace, 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@@ -1,6 +1,7 @@\n import json\n import uuid\n import hashlib\n+import six\n from flask import Flask, jsonify, request, make_response\n from localstack.services import generic_proxy\n from localstack.utils.aws import aws_stack\n@@ -140,7 +141,9 @@\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, to_bytes(kinesis_shard_id)).hex\n+    if six.PY2:\n+        kinesis_shard_id = to_bytes(kinesis_shard_id, 'utf-8')\n+    return uuid.uuid5(namespace, kinesis_shard_id).hex\n \n \n def shard_id(stream_arn, kinesis_shard_id):\n", "issue": "describe-stream crashes in docker-image\nsteps to reproduce:\r\ncreate 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\nEnable 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\ndescribe-stream using the stream arn from previous step:\r\n```\r\naws dynamodbstreams describe-stream --stream-arn arn:aws:dynamodb:eu-central-1:000000000000:table/MusicCollection/stream/2019-09-29T12:25:50.316 --endpoint http://localhost:4570\r\n```\r\ncrash:\r\n```\r\nTraceback (most recent call last):\r\n  File \"/opt/code/localstack/.venv/lib/python3.6/site-packages/flask/app.py\", line 2292, in wsgi_app\r\n    response = self.full_dispatch_request()\r\n  File \"/opt/code/localstack/.venv/lib/python3.6/site-packages/flask/app.py\", line 1815, in full_dispatch_request\r\n    rv = self.handle_user_exception(e)\r\n  File \"/opt/code/localstack/.venv/lib/python3.6/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 \"/opt/code/localstack/.venv/lib/python3.6/site-packages/flask/app.py\", line 1718, in handle_user_exception\r\n    reraise(exc_type, exc_value, tb)\r\n  File \"/opt/code/localstack/.venv/lib/python3.6/site-packages/flask/_compat.py\", line 35, in reraise\r\n    raise value\r\n  File \"/opt/code/localstack/.venv/lib/python3.6/site-packages/flask/app.py\", line 1813, in full_dispatch_request\r\n    rv = self.dispatch_request()\r\n  File \"/opt/code/localstack/.venv/lib/python3.6/site-packages/flask/app.py\", line 1799, in dispatch_request\r\n    return self.view_functions[rule.endpoint](**req.view_args)\r\n  File \"/opt/code/localstack/localstack/services/dynamodbstreams/dynamodbstreams_api.py\", line 89, in post_request\r\n    shard['ShardId'] = shard_id(stream_name, shard['ShardId'])\r\n  File \"/opt/code/localstack/localstack/services/dynamodbstreams/dynamodbstreams_api.py\", line 148, in shard_id\r\n    return '-'.join([kinesis_shard_id, random_id(stream_arn, kinesis_shard_id)])\r\n  File \"/opt/code/localstack/localstack/services/dynamodbstreams/dynamodbstreams_api.py\", line 144, in random_id\r\n    return uuid.uuid5(namespace, to_bytes(kinesis_shard_id)).hex\r\n  File \"/usr/lib/python3.6/uuid.py\", line 628, in uuid5\r\n    hash = sha1(namespace.bytes + bytes(name, \"utf-8\")).digest()\r\nTypeError: encoding without a string argument\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=None, 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        latest_stream_label = latest_stream_label or 'latest'\n        stream = {\n            'StreamArn': aws_stack.dynamodb_stream_arn(\n                table_name=table_name, 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"}], "after_files": [{"content": "import json\nimport uuid\nimport hashlib\nimport six\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=None, 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        latest_stream_label = latest_stream_label or 'latest'\n        stream = {\n            'StreamArn': aws_stack.dynamodb_stream_arn(\n                table_name=table_name, 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    if six.PY2:\n        kinesis_shard_id = to_bytes(kinesis_shard_id, 'utf-8')\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"}]}