flow-php · norberttech · Oct 27, 2025 · Oct 27, 2025 · Oct 27, 2025 · Oct 27, 2025
@@ -37,11 +37,14 @@ public function execute(InputInterface $input, OutputInterface $output) : int
 
         $paths = [
             __DIR__ . '/../src/core/etl/src/Flow/ETL/DSL/functions.php',
+            __DIR__ . '/../src/adapter/etl-adapter-avro/src/Flow/ETL/Adapter/Avro/functions.php',
             __DIR__ . '/../src/adapter/etl-adapter-chartjs/src/Flow/ETL/Adapter/ChartJS/functions.php',
             __DIR__ . '/../src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/functions.php',
             __DIR__ . '/../src/adapter/etl-adapter-doctrine/src/Flow/ETL/Adapter/Doctrine/functions.php',
+            __DIR__ . '/../src/adapter/etl-adapter-excel/src/Flow/ETL/Adapter/Excel/DSL/functions.php',
             __DIR__ . '/../src/adapter/etl-adapter-elasticsearch/src/Flow/ETL/Adapter/Elasticsearch/functions.php',
             __DIR__ . '/../src/adapter/etl-adapter-google-sheet/src/Flow/ETL/Adapter/GoogleSheet/functions.php',
+            __DIR__ . '/../src/adapter/etl-adapter-http/src/Flow/ETL/Adapter/Http/DSL/functions.php',
             __DIR__ . '/../src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/functions.php',
             __DIR__ . '/../src/adapter/etl-adapter-meilisearch/src/Flow/ETL/Adapter/Meilisearch/functions.php',
             __DIR__ . '/../src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/functions.php',

@@ -146,6 +146,7 @@
             "src/adapter/etl-adapter-elasticsearch/src/Flow/ETL/Adapter/Elasticsearch/functions.php",
             "src/adapter/etl-adapter-excel/src/Flow/ETL/Adapter/Excel/DSL/functions.php",
             "src/adapter/etl-adapter-google-sheet/src/Flow/ETL/Adapter/GoogleSheet/functions.php",
+            "src/adapter/etl-adapter-http/src/Flow/ETL/Adapter/Http/DSL/functions.php",
             "src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/functions.php",
             "src/adapter/etl-adapter-meilisearch/src/Flow/ETL/Adapter/Meilisearch/functions.php",
             "src/adapter/etl-adapter-parquet/src/Flow/ETL/Adapter/Parquet/functions.php",

@@ -61,6 +61,7 @@ parameters:
         - src/cli/src/Flow/CLI/Command/*
         - src/core/etl/src/Flow/ETL/Formatter/ASCII/ASCIITable.php
         - src/core/etl/src/Flow/ETL/Sort/ExternalSort/RowsMinHeap.php
+        - src/adapter/etl-adapter-avro/*
         - src/adapter/etl-adapter-elasticsearch/src/Flow/ETL/Adapter/Elasticsearch/ElasticsearchPHP/SearchResults.php
         - src/adapter/etl-adapter-elasticsearch/src/Flow/ETL/Adapter/Elasticsearch/ElasticsearchPHP/SearchParams.php
         - src/adapter/etl-adapter-elasticsearch/src/Flow/ETL/Adapter/Elasticsearch/ElasticsearchPHP/PointInTime.php

@@ -14,9 +14,9 @@
         __DIR__ . '/src/core/etl/src',
         __DIR__ . '/src/cli/src',
         __DIR__ . '/src/lib/*/src',
-        __DIR__ . '/src/adapter/*/src',
-        __DIR__ . '/src/bridge/*/*/src',
-        __DIR__ . '/src/tools/*/*/src',
+        __DIR__ . '/src/adapter/**/src',
+        __DIR__ . '/src/bridge/**/src',
+        __DIR__ . '/src/tools/**/src',
     ])
     ->withSkip([
         RemoveExtraParametersRector::class,
@@ -28,5 +28,5 @@
     ->withCache(__DIR__ . '/var/rector/src')
     ->withImportNames(importShortClasses: false, removeUnusedImports: true)
     ->withSets([
-        LevelSetList::UP_TO_PHP_82
-    ]);
+        LevelSetList::UP_TO_PHP_82,
+    ]);
@@ -1,5 +1,19 @@
 <?php
 
+use Flow\ETL\Adapter\Avro\FlixTech\AvroExtractor;
+use Flow\ETL\Adapter\CSV\CSVExtractor;
+use Flow\ETL\Adapter\Doctrine\DbalQueryExtractor;
+use Flow\ETL\Adapter\Elasticsearch\ElasticsearchPHP\ElasticsearchExtractor;
+use Flow\ETL\Adapter\Excel\ExcelExtractor;
+use Flow\ETL\Adapter\GoogleSheet\GoogleSheetExtractor;
+use Flow\ETL\Adapter\Http\PsrHttpClientDynamicExtractor;
+use Flow\ETL\Adapter\Http\PsrHttpClientStaticExtractor;
+use Flow\ETL\Adapter\JSON\JSONMachine\JsonExtractor;
+use Flow\ETL\Adapter\JSON\JSONMachine\JsonLinesExtractor;
+use Flow\ETL\Adapter\Meilisearch\MeilisearchPHP\MeilisearchExtractor;
+use Flow\ETL\Adapter\Parquet\ParquetExtractor;
+use Flow\ETL\Adapter\Text\TextExtractor;
+use Flow\ETL\Adapter\XML\XMLParserExtractor;
 use Flow\ETL\Config;
 use Flow\ETL\Extractor\ArrayExtractor;
 use Flow\ETL\Extractor\CacheExtractor;
@@ -67,12 +81,12 @@
         __DIR__ . '/src/core/etl/tests',
         __DIR__ . '/src/cli/tests',
         __DIR__ . '/src/lib/*/tests',
-        __DIR__ . '/src/adapter/*/tests',
-        __DIR__ . '/src/bridge/*/*/tests',
-        __DIR__ . '/src/tools/*/*/tests',
+        __DIR__ . '/src/adapter/**/tests',
+        __DIR__ . '/src/bridge/**/tests',
+        __DIR__ . '/src/tools/**/tests',
     ])
     ->withSets([
-        LevelSetList::UP_TO_PHP_82
+        LevelSetList::UP_TO_PHP_82,
     ])
     ->withRules([
         DataProviderAnnotationToAttributeRector::class,
@@ -162,6 +176,21 @@
             new NewObjectToFunction(PipelineExtractor::class, 'from_pipeline'),
             new NewObjectToFunction(DataFrameExtractor::class, 'from_data_frame'),
 
+            // Adapters
+            new NewObjectToFunction(AvroExtractor::class, 'Flow\ETL\DSL\Adapter\Avro\from_avro'),
+            new NewObjectToFunction(CSVExtractor::class, 'Flow\ETL\Adapter\CSV\from_csv'),
+            new NewObjectToFunction(DbalQueryExtractor::class, 'Flow\ETL\Adapter\Doctrine\from_dbal_query'),
+            new NewObjectToFunction(ElasticsearchExtractor::class, 'Flow\ETL\Adapter\Elasticsearch\from_es'),
+            new NewObjectToFunction(ExcelExtractor::class, 'Flow\ETL\Adapter\Excel\from_excel'),
+            new NewObjectToFunction(GoogleSheetExtractor::class, 'Flow\ETL\Adapter\GoogleSheet\from_google_sheet'),
+            new NewObjectToFunction(PsrHttpClientDynamicExtractor::class, 'Flow\ETL\Adapter\Http\from_dynamic_http_requests'),
+            new NewObjectToFunction(PsrHttpClientStaticExtractor::class, 'Flow\ETL\Adapter\Http\from_static_http_requests'),
+            new NewObjectToFunction(JsonExtractor::class, 'Flow\ETL\Adapter\JSON\from_json'),
+            new NewObjectToFunction(JsonLinesExtractor::class, 'Flow\ETL\Adapter\JSON\from_json_lines'),
+            new NewObjectToFunction(MeilisearchExtractor::class, 'Flow\ETL\Adapter\Meilisearch\from_meilisearch'),
+            new NewObjectToFunction(ParquetExtractor::class, 'Flow\ETL\Adapter\Parquet\from_parquet'),
+            new NewObjectToFunction(TextExtractor::class, 'Flow\ETL\Adapter\Text\from_text'),
+            new NewObjectToFunction(XMLParserExtractor::class, 'Flow\ETL\Adapter\XML\from_xml'),
         ]
     )
     ->withSkip([
@@ -177,4 +206,4 @@
         importDocBlockNames: true,
         importShortClasses: false,
         removeUnusedImports: true
-    );
+    );
@@ -4,31 +4,19 @@
 
 namespace Flow\ETL\DSL\Adapter\Avro;
 
-use function Flow\ETL\DSL\from_all;
 use Flow\ETL\Adapter\Avro\FlixTech\{AvroExtractor, AvroLoader};
-use Flow\ETL\{Extractor, Schema};
+use Flow\ETL\{Attribute\DocumentationDSL, Attribute\Module, Attribute\Type, Schema};
 use Flow\Filesystem\Path;
 
-function from_avro(Path|string|array $path) : Extractor
+#[DocumentationDSL(module: Module::AVRO, type: Type::EXTRACTOR)]
+function from_avro(Path|string $path) : AvroExtractor
 {
-    if (\is_array($path)) {
-        /** @var array<Extractor> $extractors */
-        $extractors = [];
-
-        foreach ($path as $next_path) {
-            $extractors[] = new AvroExtractor(
-                \is_string($next_path) ? Path::realpath($next_path) : $next_path,
-            );
-        }
-
-        return from_all(...$extractors);
-    }
-
     return new AvroExtractor(
         \is_string($path) ? Path::realpath($path) : $path
     );
 }
 
+#[DocumentationDSL(module: Module::AVRO, type: Type::LOADER)]
 function to_avro(Path|string $path, ?Schema $schema = null) : AvroLoader
 {
     return new AvroLoader(

@@ -4,9 +4,8 @@
 
 namespace Flow\ETL\Adapter\Avro\Tests\Integration;
 
-use function Flow\ETL\DSL\Adapter\Avro\to_avro;
+use function Flow\ETL\DSL\Adapter\Avro\{from_avro, to_avro};
 use function Flow\ETL\DSL\{config, flow_context};
-use Flow\ETL\Adapter\Avro\FlixTech\AvroExtractor;
 use Flow\ETL\Extractor\Signal;
 use Flow\ETL\Tests\FlowTestCase;
 use Flow\Filesystem\Path;
@@ -20,7 +19,7 @@ protected function setUp() : void
 
     public function test_limit() : void
     {
-        $extractor = new AvroExtractor(Path::realpath(__DIR__ . '/../Fixtures/orders_flow.avro'));
+        $extractor = from_avro(Path::realpath(__DIR__ . '/../Fixtures/orders_flow.avro'));
         $extractor->changeLimit(2);
 
         self::assertCount(
@@ -31,7 +30,7 @@ public function test_limit() : void
 
     public function test_signal_stop() : void
     {
-        $extractor = new AvroExtractor(Path::realpath(__DIR__ . '/../Fixtures/orders_flow.avro'));
+        $extractor = from_avro(Path::realpath(__DIR__ . '/../Fixtures/orders_flow.avro'));
 
         $generator = $extractor->extract(flow_context(config()));
 

@@ -7,7 +7,6 @@
 use function Flow\ETL\Adapter\CSV\from_csv;
 use function Flow\ETL\DSL\{df, print_schema, ref};
 use function Flow\ETL\DSL\flow_context;
-use Flow\ETL\Adapter\CSV\CSVExtractor;
 use Flow\ETL\{Config, Row, Rows, Tests\FlowTestCase};
 use Flow\ETL\Extractor\Signal;
 use Flow\Filesystem\Path;
@@ -444,7 +443,7 @@ public function test_extracting_csv_with_multiline_strings() : void
     public function test_limit() : void
     {
 
-        $extractor = new CSVExtractor(Path::realpath(__DIR__ . '/../Fixtures/orders_flow.csv'));
+        $extractor = from_csv(Path::realpath(__DIR__ . '/../Fixtures/orders_flow.csv'));
         $extractor->changeLimit(2);
 
         self::assertCount(
@@ -477,7 +476,7 @@ public function test_loading_data_from_all_partitions() : void
 
     public function test_signal_stop() : void
     {
-        $extractor = new CSVExtractor(Path::realpath(__DIR__ . '/../Fixtures/orders_flow.csv'));
+        $extractor = from_csv(Path::realpath(__DIR__ . '/../Fixtures/orders_flow.csv'));
 
         $generator = $extractor->extract(flow_context(\Flow\ETL\DSL\config()));
 

@@ -21,6 +21,7 @@ function from_excel(
     return new ExcelExtractor(\is_string($path) ? Path::realpath($path) : $path);
 }
 
+#[DocumentationDSL(module: Module::EXCEL, type: DSLType::HELPER)]
 function is_valid_excel_sheet_name(string|ScalarFunction $sheet_name) : IsValidExcelSheetName
 {
     return new IsValidExcelSheetName($sheet_name);

@@ -29,7 +29,10 @@
             "Flow\\": [
                 "src/Flow"
             ]
-        }
+        },
+        "files": [
+          "src/Flow/ETL/Adapter/Http/DSL/functions.php"
+        ]
     },
     "autoload-dev": {
         "psr-4": {

@@ -0,0 +1,37 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Adapter\Http;
+
+use Flow\ETL\{Adapter\Http\DynamicExtractor\NextRequestFactory,
+    Attribute\DocumentationDSL,
+    Attribute\Module,
+    Attribute\Type};
+use Psr\Http\Client\ClientInterface;
+use Psr\Http\Message\RequestInterface;
+
+#[DocumentationDSL(module: Module::HTTP, type: Type::EXTRACTOR)]
+function from_dynamic_http_requests(
+    ClientInterface $client,
+    NextRequestFactory $requestFactory,
+) : PsrHttpClientDynamicExtractor {
+    return new PsrHttpClientDynamicExtractor(
+        $client,
+        $requestFactory,
+    );
+}
+
+/**
+ * @param iterable<RequestInterface> $requests
+ */
+#[DocumentationDSL(module: Module::HTTP, type: Type::EXTRACTOR)]
+function from_static_http_requests(
+    ClientInterface $client,
+    iterable $requests,
+) : PsrHttpClientStaticExtractor {
+    return new PsrHttpClientStaticExtractor(
+        $client,
+        $requests,
+    );
+}
@@ -4,9 +4,9 @@
 
 namespace Flow\ETL\Adapter\HTTP\Tests\Integration;
 
+use function Flow\ETL\Adapter\Http\from_dynamic_http_requests;
 use function Flow\ETL\DSL\{config, flow_context};
 use Flow\ETL\Adapter\Http\DynamicExtractor\NextRequestFactory;
-use Flow\ETL\Adapter\Http\PsrHttpClientDynamicExtractor;
 use Flow\ETL\Tests\FlowTestCase;
 use Http\Mock\Client;
 use Nyholm\Psr7\Factory\Psr17Factory;
@@ -32,7 +32,7 @@ public function test_http_extractor() : void
             ], $fixtureContent),
         );
 
-        $extractor = new PsrHttpClientDynamicExtractor($psr18Client, new class implements NextRequestFactory {
+        $extractor = from_dynamic_http_requests($psr18Client, new class implements NextRequestFactory {
             public function create(?ResponseInterface $previousResponse = null) : ?RequestInterface
             {
                 $psr17Factory = new Psr17Factory();

@@ -4,8 +4,8 @@
 
 namespace Flow\ETL\Adapter\HTTP\Tests\Integration;
 
+use function Flow\ETL\Adapter\Http\from_static_http_requests;
 use function Flow\ETL\DSL\{config, flow_context};
-use Flow\ETL\Adapter\Http\PsrHttpClientStaticExtractor;
 use Flow\ETL\{Rows, Tests\FlowTestCase};
 use Http\Mock\Client;
 use Nyholm\Psr7\Factory\Psr17Factory;
@@ -48,7 +48,7 @@ public function test_http_extractor() : void
                 ->withHeader('User-Agent', 'flow-php/etl');
         };
 
-        $extractor = new PsrHttpClientStaticExtractor($psr18Client, $requests());
+        $extractor = from_static_http_requests($psr18Client, $requests());
 
         $rowsGenerator = $extractor->extract(flow_context(config()));
 

@@ -7,7 +7,6 @@
 use function Flow\ETL\Adapter\JSON\from_json;
 use function Flow\ETL\DSL\{data_frame, flow_context};
 use function Flow\ETL\DSL\{df, print_schema};
-use Flow\ETL\Adapter\JSON\JSONMachine\JsonExtractor;
 use Flow\ETL\{Config, Row, Rows, Tests\FlowTestCase};
 use Flow\ETL\Extractor\Signal;
 use Flow\Filesystem\Path;
@@ -105,7 +104,7 @@ public function test_extracting_json_from_local_file_stream_with_schema() : void
 
     public function test_extracting_json_from_local_file_string_uri() : void
     {
-        $extractor = new JsonExtractor(Path::realpath(__DIR__ . '/../../Fixtures/timezones.json'));
+        $extractor = from_json(Path::realpath(__DIR__ . '/../../Fixtures/timezones.json'));
 
         $total = 0;
 
@@ -132,7 +131,7 @@ public function test_extracting_json_from_local_file_string_uri() : void
 
     public function test_limit() : void
     {
-        $extractor = new JsonExtractor(\Flow\Filesystem\DSL\path(__DIR__ . '/../../Fixtures/timezones.json'));
+        $extractor = from_json(\Flow\Filesystem\DSL\path(__DIR__ . '/../../Fixtures/timezones.json'));
         $extractor->changeLimit(2);
 
         self::assertCount(
@@ -143,7 +142,7 @@ public function test_limit() : void
 
     public function test_signal_stop() : void
     {
-        $extractor = new JsonExtractor(\Flow\Filesystem\DSL\path(__DIR__ . '/../../Fixtures/timezones.json'));
+        $extractor = from_json(\Flow\Filesystem\DSL\path(__DIR__ . '/../../Fixtures/timezones.json'));
 
         $generator = $extractor->extract(flow_context(\Flow\ETL\DSL\config()));
 

@@ -7,7 +7,6 @@
 use function Flow\ETL\Adapter\JSON\from_json_lines;
 use function Flow\ETL\DSL\{data_frame, flow_context};
 use function Flow\ETL\DSL\{df, print_schema};
-use Flow\ETL\Adapter\JSON\JSONMachine\JsonLinesExtractor;
 use Flow\ETL\{Config, Row, Rows, Tests\FlowTestCase};
 use Flow\ETL\Extractor\Signal;
 use Flow\Filesystem\Path;
@@ -105,7 +104,7 @@ public function test_extracting_jsonl_from_local_file_stream_with_schema() : voi
 
     public function test_extracting_jsonl_from_local_file_string_uri() : void
     {
-        $extractor = (new JsonLinesExtractor(Path::realpath(__DIR__ . '/../../Fixtures/timezones.jsonl')));
+        $extractor = (from_json_lines(Path::realpath(__DIR__ . '/../../Fixtures/timezones.jsonl')));
 
         $total = 0;
 
@@ -132,7 +131,7 @@ public function test_extracting_jsonl_from_local_file_string_uri() : void
 
     public function test_limit() : void
     {
-        $extractor = (new JsonLinesExtractor(\Flow\Filesystem\DSL\path(__DIR__ . '/../../Fixtures/timezones.jsonl')));
+        $extractor = (from_json_lines(\Flow\Filesystem\DSL\path(__DIR__ . '/../../Fixtures/timezones.jsonl')));
         $extractor->changeLimit(2);
 
         self::assertCount(
@@ -143,7 +142,7 @@ public function test_limit() : void
 
     public function test_signal_stop() : void
     {
-        $extractor = (new JsonLinesExtractor(\Flow\Filesystem\DSL\path(__DIR__ . '/../../Fixtures/timezones.jsonl')));
+        $extractor = (from_json_lines(\Flow\Filesystem\DSL\path(__DIR__ . '/../../Fixtures/timezones.jsonl')));
 
         $generator = $extractor->extract(flow_context(\Flow\ETL\DSL\config()));