Merge pull request #25 from keboola/COM-1534-ondra

Author: ondrajodas

CONFIG.md

@@ -1,6 +1,6 @@
 # Configuring in Keboola Connection UI
 
-## Sample
+## Sample scan mode
 
 
 ```json
@@ -23,6 +23,7 @@
         "title",
         "year"
       ],
+      "mode": "scan",
       "dateFilter": {
         "field": "year",
         "format": "Y",
@@ -39,6 +40,49 @@
 }
 ```
 
+
+## Sample query mode
+
+
+```json
+{
+  "db": {
+    "endpoint": "endpoint",
+    "accessKeyId": "access key id",
+    "#secretAccessKey": "secret access key",
+    "regionName": "eu-central-1"
+  },
+  "exports": [
+    {
+      "id": 1,
+      "name": "my-movies",
+      "table": "Movies",
+      "index": "Movies_SomeIndex",
+      "enabled": true,
+      "incremental": true,
+      "primaryKey": [
+        "title",
+        "year"
+      ],
+      "mode": "query",
+      "keyConditionExpression": "#yr = :a",
+      "expressionAttributeNames": {
+        "#yr": "year"
+      },
+      "expressionAttributeValues": {
+        ":a": "2013"
+      },
+      "limit": 100,
+      "mapping": {
+        "title": "title",
+        "year": "year",
+        "info.rating": "rating"
+      }
+    }
+  ]
+}
+```
+
 ## Description of `parameters`
 
 - `db`: DynamoDB instance connection options
@@ -55,7 +99,11 @@
     - `enabled` (optional, default: `true`): if export is enabled or not (there has to be at least one enabled export)
     - `incremental`: if load of tables to storage will be incremental
     - `primaryKey`: primary key to set on imported table, defined as array
-    - `dateFilter` (optional): how to filter scanned documents
+    - `mode`: (optional): enum(scan|query) reading mode from dynamoDb - default is scan
+    - `keyConditionExpression`: (required): provide a specific value for the partition key
+    - `expressionAttributeValues`: (required): values that can be substituted in an expression
+    - `expressionAttributeNames`: (optional): substitution tokens for attribute names in an expression
+    - `dateFilter` (optional): how to filter scanned documents (only for scan mode)
         - `field`: field name in document by which you want to filter
         - `format`: date format (e.g. `Y-m-d` for date or `Y` for year)
         - `value`: date string from which date value will be created (e.g. `-2 days`)

README.md

@@ -26,26 +26,21 @@ Application is prepared for run in container, you can start development same way
 1. Clone this repository: `git clone git@github.com:keboola/dynamodb-extractor.git`
 2. Change directory: `cd dynamodb-extractor`
 3. Build services: `docker-compose build`
-4. Run tests `docker-compose run --rm app-tests` (runs `./tests.sh` script)
+4. Run tests `docker-compose run --rm app composer ci`
 
 After seeing all tests green, continue:
 
-1. Run service: `docker-compose run --rm app` (starts container with `bash`)
+1. Run service: `docker-compose run --rm app-tests` (starts container with `bash`)
 2. Create tables/indexes and load sample data: `php init.php`
 3. Write tests and code
-4. Run tests: `./tests.sh`
+4. Run tests: `composer tests`
 
 To simulate real run:
 
 1. Create data dir: `mkdir -p data`
 2. Follow configuration sample and create `config.json` file and place it to your data directory (`data/config.json`)
 3. Simulate real run (with entrypoint command): `php ./src/app.php run ./data`
 
-### Tests
-
-- all in one: `./tests.sh`
-- or separately, just check `tests.sh` file contents
-
 ## License
 
 MIT licensed, see [LICENSE](./LICENSE) file.

src/Keboola/DynamoDbExtractor/ConfigDefinition.php

@@ -6,9 +6,22 @@
 
 use Symfony\Component\Config\Definition\Builder\TreeBuilder;
 use Symfony\Component\Config\Definition\ConfigurationInterface;
+use Symfony\Component\Config\Definition\Exception\InvalidConfigurationException;
 
 class ConfigDefinition implements ConfigurationInterface
 {
+    public const MODE_SCAN = 'scan';
+
+    public const MODE_QUERY = 'query';
+
+    private const QUERY_INVALID_NODES = ['dateFilter'];
+
+    private const SCAN_INVALID_NODES = [
+        'keyConditionExpression',
+        'expressionAttributeNames',
+        'expressionAttributeValues',
+    ];
+
     public function getConfigTreeBuilder(): TreeBuilder
     {
         $treeBuilder = new TreeBuilder('parameters');
@@ -38,31 +51,45 @@ public function getConfigTreeBuilder(): TreeBuilder
                 ->end()
                 ->arrayNode('exports')
                     ->prototype('array')
+                        ->validate()->always(function ($exportItem) {
+                            if ($exportItem['mode'] === self::MODE_QUERY) {
+                                foreach (self::QUERY_INVALID_NODES as $invalidNodes) {
+                                    if (isset($exportItem[$invalidNodes])) {
+                                        throw new InvalidConfigurationException(
+                                            sprintf('Node "%s" is not allowed for query export.', $invalidNodes)
+                                        );
+                                    }
+                                }
+                            } else {
+                                foreach (self::SCAN_INVALID_NODES as $invalidNodes) {
+                                    if (isset($exportItem[$invalidNodes])) {
+                                        throw new InvalidConfigurationException(
+                                            sprintf('Node "%s" is not allowed for scan export.', $invalidNodes)
+                                        );
+                                    }
+                                }
+                            }
+                            return $exportItem;
+                        })->end()
                         ->children()
-                            ->integerNode('id')
-                                ->isRequired()
-                            ->end()
+                            ->integerNode('id')->isRequired()->end()
                             ->scalarNode('name')
                                 ->isRequired()
                                 ->cannotBeEmpty()
                             ->end()
-                            ->scalarNode('table')
-                                ->isRequired()
-                                ->cannotBeEmpty()
-                            ->end()
-                            ->scalarNode('index')
-                                ->cannotBeEmpty()
-                            ->end()
-                            ->integerNode('limit')
-                            ->end()
-                            ->variableNode('dateFilter')
-                            ->end()
-                            ->booleanNode('enabled')
-                                ->defaultValue(true)
-                            ->end()
-                            ->booleanNode('incremental')
-                                ->isRequired()
+                            ->enumNode('mode')
+                                ->values([self::MODE_SCAN, self::MODE_QUERY])
+                                ->defaultValue(self::MODE_SCAN)
                             ->end()
+                            ->scalarNode('table')->isRequired()->cannotBeEmpty()->end()
+                            ->scalarNode('index')->cannotBeEmpty()->end()
+                            ->integerNode('limit')->end()
+                            ->variableNode('dateFilter')->end()
+                            ->scalarNode('keyConditionExpression')->end()
+                            ->variableNode('expressionAttributeValues')->end()
+                            ->variableNode('expressionAttributeNames')->end()
+                            ->booleanNode('enabled')->defaultValue(true)->end()
+                            ->booleanNode('incremental')->isRequired()->end()
                             ->arrayNode('primaryKey')
                                 ->scalarPrototype()
                                 ->end()

src/Keboola/DynamoDbExtractor/Exporter.php

@@ -6,7 +6,8 @@
 
 use Aws\DynamoDb\DynamoDbClient;
 use Aws\DynamoDb\Exception\DynamoDbException;
-use Aws\DynamoDb\Marshaler;
+use Keboola\DynamoDbExtractor\ReadingAdapter\QueryReadingAdapter;
+use Keboola\DynamoDbExtractor\ReadingAdapter\ScanReadingAdapter;
 use Nette\Utils\Strings;
 use Symfony\Component\Console\Output\OutputInterface;
 use Symfony\Component\Filesystem\Filesystem;
@@ -43,42 +44,32 @@ public function __construct(
      */
     public function export(): string
     {
-        $marshaler = new Marshaler();
-
         $params = [
             'TableName' => $this->exportOptions['table'],
         ];
 
-        if (isset($this->exportOptions['index'])) {
-            $params['IndexName'] = $this->exportOptions['index'];
-        }
-
-        if (isset($this->exportOptions['dateFilter'])) {
-            $paramsFromDateFilter = $this->createParamsFromDateFilter($this->exportOptions['dateFilter']);
-            $this->consoleOutput->writeln((string) json_encode($paramsFromDateFilter));
-            $params = array_merge($params, $paramsFromDateFilter);
+        switch ($this->exportOptions['mode']) {
+            case ConfigDefinition::MODE_QUERY:
+                $readingAdapter = new QueryReadingAdapter(
+                    $this->exportOptions,
+                    $this->dynamoDbClient,
+                    $this->consoleOutput,
+                    $this->filename
+                );
+                break;
+            default:
+                $readingAdapter = new ScanReadingAdapter(
+                    $this->exportOptions,
+                    $this->dynamoDbClient,
+                    $this->consoleOutput,
+                    $this->filename
+                );
         }
-
-        $scanLimit = new ScanLimit(1000, $this->exportOptions['limit'] ?? null);
-
         try {
-            do {
-                if (isset($response, $response['LastEvaluatedKey'])) {
-                    $params['ExclusiveStartKey'] = $response['LastEvaluatedKey'];
-                }
-                $params['Limit'] = $scanLimit->getBatchSize();
-                $response = $this->dynamoDbClient->scan($params)->toArray();
-                $scanLimit->decreaseLimit($response['Count']);
-
-                /** @var array $item */
-                foreach ((array) $response['Items'] as $item) {
-                    $json = json_encode($marshaler->unmarshalItem($item));
-                    FileHelper::appendContentToFile($this->filename, $json . "\n");
-                }
-            } while ($scanLimit->shouldContinue() && isset($response['LastEvaluatedKey']));
+            $readingAdapter->read($params);
         } catch (DynamoDbException $e) {
             if ($e->getStatusCode() !== null && substr((string) $e->getStatusCode(), 0, 1) === '4') {
-                throw new UserException((string) $e->getAwsErrorCode());
+                throw new UserException((string) $e->getAwsErrorMessage());
             } else {
                 throw $e;
             }
@@ -102,22 +93,4 @@ public function cleanup(): void
     {
         $this->filesystem->remove($this->filename);
     }
-
-    /**
-     * Creates filtering params from date filter
-     */
-    private function createParamsFromDateFilter(array $dateFilter): array
-    {
-        return [
-            'FilterExpression' => '#field >= :value',
-            'ExpressionAttributeNames' => [
-                '#field' => $dateFilter['field'],
-            ],
-            'ExpressionAttributeValues' => [
-                ':value' => [
-                    'S' => date($dateFilter['format'], strtotime($dateFilter['value'])),
-                ],
-            ],
-        ];
-    }
 }

src/Keboola/DynamoDbExtractor/Parser.php

@@ -8,7 +8,6 @@
 use Keboola\CsvMap\Mapper;
 use Keboola\CsvTable\Table;
 use Nette\Utils\Strings;
-use SplFileInfo;
 use Symfony\Component\Console\Output\OutputInterface;
 use Symfony\Component\Filesystem\Filesystem;
 

src/Keboola/DynamoDbExtractor/ReadingAdapter/AbstractReadingAdapter.php

@@ -0,0 +1,44 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Keboola\DynamoDbExtractor\ReadingAdapter;
+
+use Aws\DynamoDb\DynamoDbClient;
+use Aws\DynamoDb\Marshaler;
+use Keboola\DynamoDbExtractor\FileHelper;
+use Symfony\Component\Console\Output\OutputInterface;
+
+abstract class AbstractReadingAdapter
+{
+    protected array $exportOptions;
+
+    protected DynamoDbClient $dynamoDbClient;
+
+    protected OutputInterface $consoleOutput;
+
+    protected string $filename;
+
+    public function __construct(
+        array $exportOptions,
+        DynamoDbClient $dynamoDbClient,
+        OutputInterface $consoleOutput,
+        string $filename
+    ) {
+        $this->exportOptions = $exportOptions;
+        $this->dynamoDbClient = $dynamoDbClient;
+        $this->consoleOutput = $consoleOutput;
+        $this->filename = $filename;
+    }
+
+    abstract public function read(array $params): void;
+
+    protected function saveResponseItems(Marshaler $marshaler, array $items): void
+    {
+        /** @var array $item */
+        foreach ($items as $item) {
+            $json = json_encode($marshaler->unmarshalItem($item));
+            FileHelper::appendContentToFile($this->filename, $json . "\n");
+        }
+    }
+}

src/Keboola/DynamoDbExtractor/ReadingAdapter/QueryReadingAdapter.php

@@ -0,0 +1,41 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Keboola\DynamoDbExtractor\ReadingAdapter;
+
+use Aws\DynamoDb\Marshaler;
+use Keboola\DynamoDbExtractor\ReadingLimit;
+
+class QueryReadingAdapter extends AbstractReadingAdapter
+{
+    public function read(array $params): void
+    {
+        $marshaler = new Marshaler();
+
+        if (!empty($this->exportOptions['keyConditionExpression'])) {
+            $params['KeyConditionExpression'] = $this->exportOptions['keyConditionExpression'];
+        }
+        if (!empty($this->exportOptions['expressionAttributeValues'])) {
+            $params['ExpressionAttributeValues'] = $marshaler->marshalJson(
+                (string) json_encode($this->exportOptions['expressionAttributeValues'])
+            );
+        }
+        if (!empty($this->exportOptions['expressionAttributeNames'])) {
+            $params['ExpressionAttributeNames'] = $this->exportOptions['expressionAttributeNames'];
+        }
+
+        $scanLimit = new ReadingLimit(1000, $this->exportOptions['limit'] ?? null);
+
+        do {
+            if (isset($response, $response['LastEvaluatedKey'])) {
+                $params['ExclusiveStartKey'] = $response['LastEvaluatedKey'];
+            }
+            $params['Limit'] = $scanLimit->getBatchSize();
+            $response = $this->dynamoDbClient->query($params)->toArray();
+            $scanLimit->decreaseLimit($response['Count']);
+
+            $this->saveResponseItems($marshaler, (array) $response['Items']);
+        } while ($scanLimit->shouldContinue() && isset($response['LastEvaluatedKey']));
+    }
+}