[GoogleSheetExtractor] Add support for Schema (#1862)

Author: stloyd

src/adapter/etl-adapter-google-sheet/src/Flow/ETL/Adapter/GoogleSheet/GoogleSheetExtractor.php

@@ -7,7 +7,7 @@
 use function Flow\ETL\DSL\array_to_rows;
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\Extractor\{Limitable, LimitableExtractor, Signal};
-use Flow\ETL\{Extractor, FlowContext};
+use Flow\ETL\{Extractor, FlowContext, Schema};
 use Google\Service\Sheets;
 
 final class GoogleSheetExtractor implements Extractor, LimitableExtractor
@@ -23,6 +23,8 @@ final class GoogleSheetExtractor implements Extractor, LimitableExtractor
 
     private int $rowsPerPage = 1000;
 
+    private ?Schema $schema = null;
+
     private bool $withHeader = true;
 
     public function __construct(
@@ -38,7 +40,6 @@ public function extract(FlowContext $context) : \Generator
         $cellsRange = new SheetRange($this->columnRange, 1, $this->rowsPerPage);
         $headers = [];
 
-        /** @var Sheets\ValueRange $response */
         $response = $this->service->spreadsheets_values->get(
             $this->spreadsheetId,
             $cellsRange->toString(),
@@ -107,7 +108,7 @@ function (array $rowData) use ($headers, $headersCount, $shouldPutInputIntoRows)
             $totalRows += \count($rows);
 
             foreach ($rows as $row) {
-                $signal = yield array_to_rows($row, $context->entryFactory());
+                $signal = yield array_to_rows($row, $context->entryFactory(), schema: $this->schema);
                 $this->incrementReturnedRows();
 
                 if ($signal === Signal::STOP || $this->reachedLimit()) {
@@ -163,4 +164,11 @@ public function withRowsPerPage(int $rowsPerPage) : self
 
         return $this;
     }
+
+    public function withSchema(Schema $schema) : self
+    {
+        $this->schema = $schema;
+
+        return $this;
+    }
 }

src/adapter/etl-adapter-google-sheet/tests/Flow/ETL/Adapter/GoogleSheet/Tests/Integration/GoogleSheetExtractorTest.php

@@ -4,8 +4,9 @@
 
 namespace Flow\ETL\Adapter\GoogleSheet\Tests\Integration;
 
-use function Flow\ETL\DSL\{config, flow_context};
-use Flow\ETL\Adapter\GoogleSheet\{Columns, GoogleSheetExtractor, Tests\GoogleSheetsContext};
+use function Flow\ETL\Adapter\GoogleSheet\from_google_sheet;
+use function Flow\ETL\DSL\{df, int_schema, schema, string_schema};
+use Flow\ETL\Adapter\GoogleSheet\{Tests\GoogleSheetsContext};
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\Tests\FlowTestCase;
 
@@ -18,30 +19,62 @@ protected function setUp() : void
         $this->context = new GoogleSheetsContext();
     }
 
-    public function test_extract_skip_extra_empty_rows() : void
+    public function test_extract_puts_null_in_not_matching_schema_rows() : void
     {
-        $extractor = new GoogleSheetExtractor(
-            $this->context->sheets(__DIR__ . '/../Fixtures/extra-empty-rows.json'),
-            '1234567890',
-            new Columns('Sheet', 'A', 'Z'),
-        );
+        $rows = df()
+            ->extract(
+                from_google_sheet(
+                    $this->context->sheets(__DIR__ . '/../Fixtures/extra-empty-rows.json'),
+                    '1234567890',
+                    'Sheet',
+                )->withSchema(
+                    schema(
+                        string_schema('Header 1'),
+                        string_schema('Header 2'),
+                        int_schema('id'),
+                    )
+                )
+            )
+            ->fetch()
+            ->toArray();
 
-        $rows = $extractor->extract(flow_context(config()));
+        foreach ($rows as $row) {
+            self::assertNotSame([], $row);
+            self::assertArrayNotHasKey('Header 3', $row);
+            self::assertNull($row['id']);
+        }
+    }
+
+    public function test_extract_skip_extra_empty_rows() : void
+    {
+        $rows = df()
+            ->extract(
+                from_google_sheet(
+                    $this->context->sheets(__DIR__ . '/../Fixtures/extra-empty-rows.json'),
+                    '1234567890',
+                    'Sheet',
+                )
+            )
+            ->fetch()
+            ->toArray();
 
         foreach ($rows as $row) {
-            self::assertNotSame([], $row->toArray());
+            self::assertNotSame([], $row);
         }
     }
 
     public function test_extract_with_cut_extra_columns() : void
     {
-        $extractor = new GoogleSheetExtractor(
-            $this->context->sheets(__DIR__ . '/../Fixtures/extra-columns.json'),
-            '1234567890',
-            new Columns('Sheet', 'A', 'Z'),
-        );
-
-        $rows = $extractor->extract(flow_context(config()));
+        $rows = df()
+            ->extract(
+                from_google_sheet(
+                    $this->context->sheets(__DIR__ . '/../Fixtures/extra-columns.json'),
+                    '1234567890',
+                    'Sheet',
+                )
+            )
+            ->fetch()
+            ->toArray();
 
         foreach ($rows as $row) {
             self::assertNotNull($row);
@@ -50,20 +83,18 @@ public function test_extract_with_cut_extra_columns() : void
 
     public function test_extract_without_cut_extra_columns() : void
     {
-        $extractor = new GoogleSheetExtractor(
-            $this->context->sheets(__DIR__ . '/../Fixtures/extra-columns.json'),
-            '1234567890',
-            new Columns('Sheet', 'A', 'Z'),
-        );
-        $extractor->withDropExtraColumns(false);
-
-        $rows = $extractor->extract(flow_context(config()));
-
         $this->expectException(InvalidArgumentException::class);
         $this->expectExceptionMessage('Row has more columns (4) than headers (3)');
 
-        foreach ($rows as $row) {
-            self::assertNotNull($row);
-        }
+        df()
+            ->extract(
+                from_google_sheet(
+                    $this->context->sheets(__DIR__ . '/../Fixtures/extra-columns.json'),
+                    '1234567890',
+                    'Sheet',
+                )->withDropExtraColumns(false)
+            )
+            ->fetch()
+            ->toArray();
     }
 }