[ExcelExtractor] Add support for Schema (#1864)

stloyd · web-flow · commit 0b878f3780c7 · 2025-09-24T20:04:43.000+02:00
* [ExcelExtractor] Add support for `Schema`

* [ExcelExtractor] Add support for `Schema`
diff --git a/src/adapter/etl-adapter-excel/src/Flow/ETL/Adapter/Excel/ExcelExtractor.php b/src/adapter/etl-adapter-excel/src/Flow/ETL/Adapter/Excel/ExcelExtractor.php
@@ -9,8 +9,8 @@
     Adapter\Excel\Sheet\SheetsManager,
     Exception\InvalidArgumentException,
     Extractor,
-    FlowContext
-};
+    FlowContext,
+    Schema};
 use Flow\ETL\Extractor\{FileExtractor, Limitable, LimitableExtractor, PathFiltering, Signal};
 use Flow\Filesystem\{Path, SourceStream};
 use OpenSpout\Common\Entity\{Cell, Row};
@@ -28,6 +28,8 @@ final class ExcelExtractor implements Extractor, FileExtractor, LimitableExtract
 
     private XlsxReader|OdsReader|null $reader = null;
 
+    private ?Schema $schema = null;
+
     private ?string $sheetName = null;
 
     private bool $withHeader = true;
@@ -57,8 +59,7 @@ public function extract(FlowContext $context) : \Generator
         foreach ($context->streams()->list($this->path, $this->filter()) as $stream) {
             foreach ($this->extractRows($stream, $headers, $offset) as $row) {
                 // Ensure $row is an array before passing to array_to_rows
-                $rowArray = \is_array($row) ? $row : [];
-                $signal = yield array_to_rows($rowArray, $context->entryFactory(), $stream->path()->partitions());
+                $signal = yield array_to_rows(\is_array($row) ? $row : [], $context->entryFactory(), $stream->path()->partitions(), schema: $this->schema);
                 $this->incrementReturnedRows();
 
                 if ($signal === Signal::STOP || $this->reachedLimit()) {
@@ -112,6 +113,13 @@ public function withReader(ExcelReader $reader) : self
         return $this;
     }
 
+    public function withSchema(Schema $schema) : self
+    {
+        $this->schema = $schema;
+
+        return $this;
+    }
+
     public function withSheetName(string $sheetName) : self
     {
         SheetNameAssertion::assert($sheetName);
diff --git a/src/adapter/etl-adapter-excel/tests/Flow/ETL/Adapter/Excel/Tests/Integration/ExcelExtractorTest.php b/src/adapter/etl-adapter-excel/tests/Flow/ETL/Adapter/Excel/Tests/Integration/ExcelExtractorTest.php
@@ -5,10 +5,10 @@
 namespace Flow\ETL\Adapter\Excel\Tests\Integration;
 
 use function Flow\ETL\Adapter\Excel\DSL\from_excel;
-use function Flow\ETL\DSL\{config, df, flow_context};
+use function Flow\ETL\DSL\{config, df, flow_context, int_schema, schema, string_schema};
 use Flow\ETL\Adapter\Excel\ExcelReader;
 use Flow\ETL\Exception\InvalidArgumentException;
-use Flow\ETL\{Extractor\Signal, Row, Rows};
+use Flow\ETL\{Extractor\Signal, Rows};
 use Flow\ETL\Tests\FlowTestCase;
 use Flow\Filesystem\{Partition, Path};
 use PHPUnit\Framework\Attributes\DataProvider;
@@ -68,19 +68,17 @@ public function test_extract_excel_file_with_limit(string $fixtureName) : void
         $extractor = from_excel($fixtureName);
         $extractor->changeLimit(5);
 
-        $total = 0;
+        $rows = df()
+            ->extract($extractor)
+            ->fetch()
+            ->toArray();
 
-        foreach ($extractor->extract(flow_context(config())) as $rows) {
-            $rows->each(function (Row $row) : void {
-                $this->assertSame(
-                    ['id', 'name', 'email'],
-                    \array_keys($row->toArray())
-                );
-            });
-            $total += $rows->count();
-        }
+        self::assertCount(5, $rows);
 
-        self::assertSame(5, $total);
+        foreach ($rows as $row) {
+            self::assertSame(['id', 'name', 'email'], \array_keys($row));
+            self::assertCount(3, $row);
+        }
     }
 
     #[DataProvider('provide_fixtures')]
@@ -89,19 +87,17 @@ public function test_extract_excel_file_with_offset(string $fixtureName) : void
         $extractor = from_excel($fixtureName);
         $extractor->withOffset(5);
 
-        $total = 0;
+        $rows = df()
+            ->extract($extractor)
+            ->fetch()
+            ->toArray();
 
-        foreach ($extractor->extract(flow_context(config())) as $rows) {
-            $rows->each(function (Row $row) : void {
-                $this->assertSame(
-                    ['id', 'name', 'email'],
-                    \array_keys($row->toArray())
-                );
-            });
-            $total += $rows->count();
-        }
+        self::assertCount(7, $rows);
 
-        self::assertSame(7, $total);
+        foreach ($rows as $row) {
+            self::assertSame(['id', 'name', 'email'], \array_keys($row));
+            self::assertCount(3, $row);
+        }
     }
 
     #[DataProvider('provide_fixtures')]
@@ -111,112 +107,135 @@ public function test_extract_excel_file_with_offset_without_header(string $fixtu
         $extractor->withHeader(false);
         $extractor->withOffset(5);
 
-        $total = 0;
+        $rows = df()
+            ->extract($extractor)
+            ->fetch()
+            ->toArray();
 
-        foreach ($extractor->extract(flow_context(config())) as $rows) {
-            $rows->each(function (Row $row) : void {
-                $this->assertSame(
-                    ['e00', 'e01', 'e02'],
-                    \array_keys($row->toArray())
-                );
-            });
-            $total += $rows->count();
-        }
+        self::assertCount(6, $rows);
 
-        self::assertSame(6, $total);
+        foreach ($rows as $row) {
+            self::assertSame(['e00', 'e01', 'e02'], \array_keys($row));
+            self::assertCount(3, $row);
+        }
     }
 
     #[DataProvider('provide_fixtures')]
     public function test_extract_excel_file_with_selected_sheet_name(string $fixtureName) : void
     {
-        $extractor = from_excel($fixtureName);
-        $extractor->withSheetName('Sheet2');
+        $rows = df()
+            ->extract(
+                from_excel($fixtureName)
+                    ->withSheetName('Sheet2')
+            )
+            ->fetch()
+            ->toArray();
 
-        $total = 0;
+        self::assertCount(5, $rows);
 
-        foreach ($extractor->extract(flow_context(config())) as $rows) {
-            $rows->each(function (Row $row) : void {
-                $this->assertSame(
-                    ['id', 'name', 'email'],
-                    \array_keys($row->toArray())
-                );
-            });
-            $total += $rows->count();
+        foreach ($rows as $row) {
+            self::assertSame(['id', 'name', 'email'], \array_keys($row));
+            self::assertCount(3, $row);
         }
-
-        self::assertSame(5, $total);
     }
 
     #[DataProvider('provide_fixtures')]
     public function test_extract_excel_file_with_unknown_sheet_name(string $fixtureName) : void
     {
-        $extractor = from_excel($fixtureName);
-        $extractor->withSheetName('unknown');
-
         $this->expectException(InvalidArgumentException::class);
         $this->expectExceptionMessage("Sheet with name: 'unknown' not found.");
 
-        iterator_to_array($extractor->extract(flow_context(config())));
+        df()
+            ->extract(
+                from_excel($fixtureName)
+                    ->withSheetName('unknown')
+            )
+            ->fetch()
+            ->toArray();
     }
 
     #[DataProvider('provide_fixtures')]
     public function test_extract_excel_file_without_header(string $fixtureName) : void
     {
-        $extractor = from_excel($fixtureName);
-        $extractor->withHeader(false);
+        $rows = df()
+            ->extract(
+                from_excel($fixtureName)
+                    ->withHeader(false)
+            )
+            ->fetch()
+            ->toArray();
 
-        $total = 0;
+        self::assertCount(10, $rows);
 
-        foreach ($extractor->extract(flow_context(config())) as $rows) {
-            $rows->each(function (Row $row) : void {
-                $this->assertSame(
-                    ['e00', 'e01', 'e02'],
-                    \array_keys($row->toArray())
-                );
-            });
-            $total += $rows->count();
+        foreach ($rows as $row) {
+            self::assertSame(['e00', 'e01', 'e02'], \array_keys($row));
+            self::assertCount(3, $row);
         }
-
-        self::assertSame(10, $total);
     }
 
     #[DataProvider('provide_nullable_fixtures')]
     public function test_extract_excel_nullable_file(string $fixtureName) : void
     {
-        $extractor = from_excel($fixtureName);
+        $rows = df()
+            ->extract(from_excel($fixtureName))
+            ->fetch()
+            ->toArray();
 
-        $total = 0;
+        self::assertCount(5, $rows);
 
-        foreach ($extractor->extract(flow_context(config())) as $rows) {
-            $rows->each(function (Row $row) : void {
-                $this->assertSame(['id', 'name', 'email'], \array_keys($row->toArray()));
-                $this->assertCount(3, $row->toArray());
-            });
-            $total += $rows->count();
+        foreach ($rows as $row) {
+            self::assertSame(['id', 'name', 'email'], \array_keys($row));
+            self::assertCount(3, $row);
         }
+    }
 
-        self::assertSame(5, $total);
+    #[DataProvider('provide_fixtures')]
+    public function test_extract_excel_puts_null_in_not_matching_schema_rows(string $fixtureName) : void
+    {
+        $rows = df()
+            ->extract(
+                from_excel($fixtureName)
+                    ->withSchema(
+                        schema(
+                            int_schema('id'),
+                            string_schema('name'),
+                            string_schema('email'),
+                            string_schema('missing'),
+                        )
+                    )
+            )
+            ->fetch()
+            ->toArray();
+
+        foreach ($rows as $row) {
+            self::assertNotSame([], $row);
+            self::assertNull($row['missing']);
+        }
     }
 
     public function test_extract_with_unknown_file() : void
     {
-        $extractor = from_excel(__DIR__ . '/../Fixtures/empty_file');
-
         $this->expectException(InvalidArgumentException::class);
         $this->expectExceptionMessage('Unsupported file format: n/a');
 
-        iterator_to_array($extractor->extract(flow_context(config())));
+        df()
+            ->extract(from_excel(__DIR__ . '/../Fixtures/empty_file'))
+            ->fetch()
+            ->toArray();
     }
 
     public function test_extract_with_wrongly_selected_reader() : void
     {
-        $extractor = from_excel(__DIR__ . '/../Fixtures/fixture.xlsx');
-        $extractor->withReader(ExcelReader::ODS);
-
         $this->expectException(InvalidArgumentException::class);
         $this->expectExceptionMessage('Failed to open file: Could not open');
 
-        iterator_to_array($extractor->extract(flow_context(config())));
+        df()
+            ->extract(
+                from_excel(__DIR__ . '/../Fixtures/fixture.xlsx')
+                    ->withReader(ExcelReader::ODS)
+            )
+            ->fetch()
+            ->toArray();
     }
 
     public function test_loading_data_from_all_partitions() : void
