updates to custom migration path

Author: robm26

.gitignore

@@ -20,8 +20,10 @@ package-lock.json
 **/.DS_Store
 **/.chalice/deployments/
 **/.chalice/deployed/
-target-tables/
-relational-migration/syncs3.sh
-.chalice/config.json
-syncs3.sh
+workshops/relational-migration/target-tables/
+workshops/relational-migration/syncs3.sh
+workshops/relational-migration/.chalice/config.json
+workshops/relational-migration/setenv.sh
+workshops/relational-migration/setenvd.sh
+workshops/relational-migration/source-tables/app_db.*
 

workshops/relational-migration/.chalice/relational-migration-policy.json

@@ -0,0 +1,52 @@
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Sid": "DynamoDBIndexAccess",
+      "Effect": "Allow",
+      "Action": [
+        "dynamodb:Scan",
+        "dynamodb:Query"
+      ],
+      "Resource": [
+        "arn:aws:dynamodb:*:*:table/*/index/*"
+      ]
+    },
+    {
+      "Sid": "DynamoDBTableAccess",
+      "Effect": "Allow",
+      "Action": [
+        "dynamodb:BatchGetItem",
+        "dynamodb:BatchWriteItem",
+        "dynamodb:ConditionCheckItem",
+        "dynamodb:PutItem",
+        "dynamodb:DescribeTable",
+        "dynamodb:DeleteItem",
+        "dynamodb:GetItem",
+        "dynamodb:Scan",
+        "dynamodb:Query",
+        "dynamodb:UpdateItem"
+      ],
+      "Resource": "arn:aws:dynamodb:*:*:table/*"
+    },
+    {
+      "Sid": "DynamoDBAccount",
+      "Effect": "Allow",
+      "Action": [
+        "dynamodb:ListTables"
+      ],
+      "Resource": "*"
+    },
+    {
+      "Sid": "LambdaIAMLogs",
+      "Effect": "Allow",
+      "Action": [
+        "logs:CreateLogGroup",
+        "logs:CreateLogStream",
+        "logs:PutLogEvents",
+        "logs:DescribeLogStreams"
+      ],
+      "Resource": ["arn:aws:logs:*:*:*"]
+    }
+  ]
+}

workshops/relational-migration/README.md

@@ -1,5 +1,5 @@
 # Relational Database to DynamoDB Migration: Workshop Artifacts
 
 Here you will find scripts and a sample database application
-to accompany the Relational Database Migration Workshop hands-on instructions
+to accompany the Relational Database Migration Workshop hands-on instructions 
 published at [catalog.workshops.aws](https://catalog.workshops.aws/)

workshops/relational-migration/app.py

@@ -1,33 +1,48 @@
+from chalice import Chalice
+import boto3
+from botocore.exceptions import ClientError
+import logging
 import os
-# import time, datetime
 import json
-from chalice import Chalice
-from chalicelib import mysql_calls as db
-# from chalicelib import dynamodb_calls as db
-
 import mysql.connector
-import logging
 
-import boto3
-from botocore.exceptions import ClientError
+
+migration_stage = 'dynamodb'
+# 'relational', 'dual-write', 'dynamodb']
+
+if "MIGRATION_STAGE" in os.environ:
+    migration_stage = os.environ['MIGRATION_STAGE']
+
+if migration_stage == 'relational' or migration_stage == 'dual-write':
+    from chalicelib import mysql_calls as db
+else:
+    from chalicelib import dynamodb_calls as db
 
 app = Chalice(app_name='migration')
 
 region = "us-east-2"
-sql = "select * from Boston"
 
 if "AWS_DEFAULT_REGION" in os.environ:
     region = os.environ['AWS_DEFAULT_REGION']
 
 @app.route('/', methods=['GET'], cors=True)
 def ping():
-    return {'engine': db.engine()}
+    request = app.current_request
+    context = request.to_dict()['context']
+    return_status = {'engine': db.engine()}
+
+    if 'stage' in context:
+        return_status['stage'] = context['stage']
+
+    return return_status
 
 
 @app.route('/list_tables', methods=['GET'], cors=True)
 def list_tables():
     request = app.current_request
-#     print(request.to_dict())
+#     print('*****')
+#     print(json.dumps(request.to_dict(), indent=2))
+
     return db.list_tables()
 
 

workshops/relational-migration/chalicelib/dynamodb_calls.py

@@ -1,6 +1,200 @@
 # Implementation of application's DynamoDB database calls
+import boto3
+from boto3.dynamodb.types import TypeSerializer, TypeDeserializer
+from botocore.config import Config
+from botocore.exceptions import ClientError
+
+import os
+import time, datetime
+import json
+
+deserializer = boto3.dynamodb.types.TypeDeserializer()
+serializer = boto3.dynamodb.types.TypeSerializer()
+
+return_limit = 20
+region = 'us-east-2'
+
+ddb = boto3.client('dynamodb', region_name=region)
+ddbr = boto3.resource('dynamodb', region_name=region)
+# resource vs client: https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/programming-with-python.html#programming-with-python-client-resource
+
+serializer = TypeSerializer()
+deserializer = TypeDeserializer()
+table_metadata = {}
+
+
+def engine():
+    return "DynamoDB"
 
 
 def list_tables():
+    response = None
+    table_names = []
+    try:
+        response = ddb.list_tables()
+        # print('ddb list_tables response')
+        base_tables = response['TableNames']
+        for table in base_tables:
+            table_names.append(table)
+
+            response_dt = ddb.describe_table(TableName=table)
+            table_metadata = json.loads(json.dumps(response_dt['Table'], default=serialize_datetime))
+
+            if 'GlobalSecondaryIndexes' in table_metadata:
+                gsis = table_metadata['GlobalSecondaryIndexes']
+                for gsi in gsis:
+                    table_names.append(table + '.' + gsi['IndexName'])
+
+    except Exception as err:
+        return {'Error': str(err)}
+
+    return table_names
+
+
+def desc_table(table):
+    try:
+        response = ddb.describe_table(TableName=table)
+        response_str = json.dumps(response['Table'], default=serialize_datetime)
+
+    except Exception as err:
+        return {'Error': str(err)}
+
+    # table_metadata[table] = json.loads(response_str)
+
+    return response_str
+
+
+def scan_table(table):
+    response = None
+    key_list = get_keys(table)
+
+    try:
+        response = ddb.scan(TableName=table)
+    except Exception as err:
+        return {'Error': str(err)}
+
+    items = []
+
+    ddb_items = response['Items']
+
+    for item in ddb_items:
+        new_item = deserialize_ddb(item) # convert DynamoDB JSON to plain JSON
+        items.append(new_item)
+
+    return items
+
+
+def query(table, request):
+
+    keyList = list(request['queryRequest']['queryConditions'].keys())
+    sql_condition = keyList[0] + ' = %s'
+
+    if len(keyList) > 1:
+        sql_condition += ' AND ' + keyList[1] + ' = %s'
+
+    key_vals = list(request['queryRequest']['queryConditions'].values())
+
+    query_stmt = 'SELECT * FROM ' + table + ' WHERE ' + sql_condition
+
+
+#     mysql_cur.execute(query_stmt, key_vals)
+#     result = mysql_cur.fetchall()
+#     dataset = format_sql_dataset(result)
+
+    return dataset
+
+
+def get_record(table, request):
+    get_request = {'TableName': table}
+    keyList = list(request['Key'].keys())
+    pk_name = keyList[0]
+    pk_value = request['Key'][keyList[0]]
+    sk_name = None
+    sk_value = None
+    if len(keyList) > 1:
+        sk_name = keyList[1]
+        sk_value = request['Key'][keyList[1]]
+
+    response = None
+    key_list = get_keys(table)
+    get_request['Key'] = request['Key']
+
+    try:
+        table = ddbr.Table(table)
+        response = table.get_item(**get_request)
+        # print('response: ')
+        # print(response['Item'])
+
+        return response['Item']
+    except Exception as err:
+        return {'Error': str(err)}
+
+    items = []
+
+#     ddb_items = response['Items']
+#     print(ddb_items)
+
+    return response['Item']
+
+
+def new_record(table, record):
+
+    return({"status":1})
+
+
+def update_record(table, request):
+
+    return({"status": 1})
+
+
+def delete_record(table, recordKey):
+
+    return({"status":1})
+
+
+def serialize_datetime(obj):
+    if isinstance(obj, datetime.datetime):
+        formatted = obj.isoformat()
+        return formatted[:19].replace('T', ' ')
+    raise TypeError("Type not serializable")
+
+def deserialize_ddb(dynamodb_json_string):
+    return deserializer.deserialize({'M': dynamodb_json_string})
+
+def get_keys(table):
+    key_list = []
+    if table not in table_metadata:
+        try:
+            response = ddb.describe_table(TableName=table)
+            response_str = json.dumps(response['Table'], default=serialize_datetime)
+
+        except Exception as err:
+            return {'Error': str(err)}
+
+        table_metadata[table] = json.loads(response_str)
+
+    key_schema = table_metadata[table]['KeySchema']
+
+    hash_key = [key for key in key_schema if key['KeyType'] == 'HASH']
+    range_key = [key for key in key_schema if key['KeyType'] == 'RANGE']
+
+    key_list.append(hash_key[0]['AttributeName'])
+    if range_key:
+        key_list.append(range_key[0]['AttributeName'])
+
+    return key_list
+
+
+def dynamo_to_python(dynamo_object: dict) -> dict:
+    deserializer = TypeDeserializer()
+    return {
+        k: deserializer.deserialize(v)
+        for k, v in dynamo_object.items()
+    }
 
-    return ['Table1','Table2','Table3']
+def python_to_dynamo(python_object: dict) -> dict:
+    serializer = TypeSerializer()
+    return {
+        k: serializer.serialize(v)
+        for k, v in python_object.items()
+    }

workshops/relational-migration/chalicelib/mysql_calls.py

@@ -43,10 +43,12 @@ def engine():
 
 
 def list_tables():
-    request = "SELECT TABLE_NAME FROM information_schema.tables WHERE table_schema = '" + mysql_db + "'"
+    request = "SELECT TABLE_NAME FROM information_schema.tables "
+    request += "WHERE table_schema = '" + mysql_db + "' AND table_type = 'BASE TABLE'"
     mysql_cur.execute(request)
     result = mysql_cur.fetchall()
-    print(result)
+    # print(result)
+
     tables = []
     for t in result:
         tables.append(t['TABLE_NAME'])
@@ -97,7 +99,6 @@ def scan_table(table):
     request += "FROM " + mysql_db + "." + table + " "
     request += "LIMIT " + str(limit)
 
-
     mysql_cur.execute(request)
     result = mysql_cur.fetchall()
     dataset = format_sql_dataset(result)
@@ -115,9 +116,9 @@ def query(table, request):
 
     key_vals = list(request['queryRequest']['queryConditions'].values())
 
-    get_stmt = 'SELECT * FROM ' + table + ' WHERE ' + sql_condition
+    query_stmt = 'SELECT * FROM ' + table + ' WHERE ' + sql_condition
 
-    mysql_cur.execute(get_stmt, key_vals)
+    mysql_cur.execute(query_stmt, key_vals)
     result = mysql_cur.fetchall()
     dataset = format_sql_dataset(result)
 
@@ -126,13 +127,13 @@ def query(table, request):
 
 def get_record(table, request):
 
-    keyList = list(request['recordKey'].keys())
+    keyList = list(request['Key'].keys())
     sql_condition = keyList[0] + ' = %s'
 
     if len(keyList) > 1:
         sql_condition += ' AND ' + keyList[1] + ' = %s'
 
-    key_vals = list(request['recordKey'].values())
+    key_vals = list(request['Key'].values())
 
     get_stmt = 'SELECT * FROM ' + table + ' WHERE ' + sql_condition
 
@@ -207,6 +208,7 @@ def delete_record(table, recordKey):
     return({"status":mysql_cur.rowcount})
 
 def format_sql_dataset(dataset):
+
     formatted_dataset = []
     for index, row in enumerate(dataset):
         formatted_row = {}