feat: initial working version of dynamodb driver

packages/data-migration-cli/README.md

@@ -95,7 +95,7 @@ USAGE
 
 OPTIONS
   -h, --help       show CLI help
-  --config=config  [default: /Users/bforce/repos/data-migration/packages/data-migration-cli/.swallow.js]
+  --config=config  [default: /Users/bforce/repos/data-migration/packages/data-migration-cli/.dm.config.js]
 ```
 
 _See code: [src/commands/new.ts](https://github.com/theBenForce/data-migration/blob/v1.0.18/src/commands/new.ts)_

packages/data-migration/src/DriverBuilder.ts

@@ -1,8 +1,5 @@
 import { Driver } from "./DriverTypes";
 
-type DriverBuilder = <T>(
-  params: T,
-  logger: (message: string) => void
-) => Driver;
+type DriverBuilder<T> = (params: T, logger: (message: string) => void) => Driver;
 
 export default DriverBuilder;

packages/data-migration/src/DriverTypes/RDS.ts

@@ -1,5 +1,6 @@
 import Driver from "./Driver";
+import { Observable } from "rxjs";
 
 export default interface RDS extends Driver {
-  query: <T>(query: string, params: Array<any>) => Promise<Array<T>>;
+  query<T>(query: string, params: Array<any>): Observable<T>;
 }

packages/data-migration/src/index.ts

@@ -2,11 +2,7 @@ import DriverBuilder from "./DriverBuilder";
 
 export { default as Processor } from "./Processor";
 export { ProcessorParams } from "./Processor";
-export {
-  default as MigrationScript,
-  ScriptContext,
-  MigrationExecutor
-} from "./MigrationScript";
+export { default as MigrationScript, ScriptContext, MigrationExecutor } from "./MigrationScript";
 export { default as DriverBuilder } from "./DriverBuilder";
 export { default as Configuration } from "./Config";
 export { Driver } from "./DriverTypes";
@@ -15,7 +11,7 @@ import * as _ from "lodash";
 
 export interface LoadConfigParameters<T> {
   [key: string]: {
-    driver: DriverBuilder;
+    driver: DriverBuilder<any>;
     params: {
       [key: string]: T;
     };

plugins/drivers/aurora-rds/package.json

@@ -10,7 +10,8 @@
   },
   "dependencies": {
     "aws-sdk": "^2.630.0",
-    "data-migration": "^1.0.18"
+    "data-migration": "^1.0.18",
+    "rxjs": "^6.5.4"
   },
   "scripts": {
     "build": "tsc",

plugins/drivers/aurora-rds/src/index.ts

@@ -1,20 +1,16 @@
 import * as AWS from "aws-sdk";
-// tslint:disable-next-line: no-duplicate-imports
-import { RDSDataService } from "aws-sdk";
-import {
-  ColumnMetadata,
-  ExecuteStatementRequest
-} from "aws-sdk/clients/rdsdataservice";
+import { ColumnMetadata, ExecuteStatementRequest } from "aws-sdk/clients/rdsdataservice";
 import { DriverBuilder } from "data-migration";
 import RDSDriver from "data-migration/lib/DriverTypes/RDS";
+import { Observable } from "rxjs";
 
 function convertResultsToObject<T>(
   metadata?: AWS.RDSDataService.ColumnMetadata[]
-): (record: RDSDataService.Field[]) => T {
+): (record: AWS.RDSDataService.Field[]) => T {
   if (!metadata) {
     throw new Error(`No metadata defined!`);
   }
-  return (record: RDSDataService.Field[]) => {
+  return (record: AWS.RDSDataService.Field[]) => {
     const result: { [key: string]: any } = {};
 
     metadata.forEach((value: ColumnMetadata, idx: number) => {
@@ -34,68 +30,69 @@ const rdsDriver: DriverBuilder<Record<string, string>> = (
   params: { [key: string]: string },
   logger: (message: string) => void
 ): RDSDriver => {
-  let dataService: RDSDataService;
-  let transactionId: string;
+  let dataService: AWS.RDSDataService;
+  let transactionId: string | undefined;
   let paramsBase = {
     resourceArn: params.resourceArn,
-    secretArn: params.secretArn
+    secretArn: params.secretArn,
   };
 
   return {
-    async query<T>(
-      query: string,
-      parameters: Array<RDSDataService.SqlParameter>
-    ): Promise<Array<T>> {
-      const queryParameters: ExecuteStatementRequest = {
-        ...paramsBase,
-        sql: query,
-        database: params.database,
-        schema: params.schema,
-        includeResultMetadata: true,
-        transactionId,
-        parameters
-      };
+    query<T>(query: string, parameters: Array<AWS.RDSDataService.SqlParameter>): Observable<T> {
+      // @ts-ignore
+      return new Observable<T>(async (subscriber) => {
+        const queryParameters: ExecuteStatementRequest = {
+          ...paramsBase,
+          sql: query,
+          database: params.database,
+          schema: params.schema,
+          includeResultMetadata: true,
+          transactionId,
+          parameters,
+        };
 
-      const result = await dataService
-        .executeStatement(queryParameters)
-        .promise();
+        const result = await dataService.executeStatement(queryParameters).promise();
 
-      if (result.records === undefined) {
-        return [];
-      }
+        if (result.records === undefined) {
+          return [];
+        }
+
+        result.records
+          .map(convertResultsToObject<T>(result.columnMetadata))
+          .forEach((record) => subscriber.next(record));
 
-      return result.records.map(
-        convertResultsToObject<T>(result.columnMetadata)
-      );
+        subscriber.complete();
+      });
     },
 
     async init() {
       dataService = new AWS.RDSDataService({
         apiVersion: "2018-08-01",
-        region: params.region
+        region: params.region,
       });
 
       const transactionParams = {
         ...paramsBase,
         database: params.database,
-        schema: params.schema
+        schema: params.schema,
       };
 
       logger(`Creating transaction`);
-      let { transactionId } = await dataService
-        .beginTransaction(transactionParams)
-        .promise();
+      const result = await dataService.beginTransaction(transactionParams).promise();
+      transactionId = result.transactionId;
     },
 
     async cleanup() {
-      const transactionParams = {
-        ...paramsBase,
-        transactionId
-      };
+      if (transactionId !== undefined) {
+        const transactionParams = {
+          ...paramsBase,
+          transactionId,
+        };
 
-      logger(`Committing transaction`);
-      await dataService.commitTransaction(transactionParams).promise();
-    }
+        logger(`Committing transaction`);
+        await dataService.commitTransaction(transactionParams).promise();
+      }
+    },
   };
 };
 

plugins/drivers/dynamodb/README.md

@@ -0,0 +1,49 @@
+# Welcome to dm-driver-dynamodb 👋
+
+[![Build Status](https://travis-ci.org/theBenForce/data-migration.svg?branch=master)](https://travis-ci.org/theBenForce/data-migration)
+[![NPM Package](https://img.shields.io/npm/v/dm-driver-dynamodb)](https://www.npmjs.com/package/dm-driver-dynamodb)
+[![Maintainability](https://api.codeclimate.com/v1/badges/89a0c1976c9b89979635/maintainability)](https://codeclimate.com/github/theBenForce/data-migration/maintainability)
+[![lerna](https://img.shields.io/badge/maintained%20with-lerna-cc00ff.svg)](https://lerna.js.org/)
+
+> A [Data Migration](https://www.npmjs.com/package/data-migration) driver to load/create DynamoDB entries.
+
+# Configuration
+
+## Parameters
+
+The DynamoDB driver accepts the following parameters as part of its configuration:
+
+| Name            | Type   | Required | Description                                                                            |
+| --------------- | ------ | -------- | -------------------------------------------------------------------------------------- |
+| region          | string | Yes      | The AWS Region where this table exists                                                 |
+| TableName       | string | Yes      | The name of the DynamoDB table to connect to                                           |
+| accessKeyId     | string | No       | AWS Credentials, if not provided data-migration will use the default AWS configuration |
+| secretAccessKey | string | No       |
+
+## Sample Configuration
+
+```javascript
+module.exports = {
+  defaultStage: "prod",
+  migrationDirectory: "migrations",
+  stages: {
+    prod: {
+      users: {
+        driver: require("dm-driver-dynamodb"),
+        params: {
+          region: "us-east-1",
+          TableName: {
+            // Use this processor to get values from CloudFormation
+            processor: require("dm-processor-cf"),
+            params: {
+              stack: "some-stack-name",
+              output: "SomeOutputName",
+              region: "us-east-1",
+            },
+          },
+        },
+      },
+    },
+  },
+};
+```