Doctrine Dbal Transactional Loader (#1843)

* Doctrine Dbal Transactional Loader

* Updated DSL definitions

* Fixed tests failing at lowest version of dependencies

* Updated landing page dependencies, fixed failing build

Author: norberttech

composer.lock

@@ -97,7 +97,7 @@ public function extract(FlowContext $context) : \Generator
                     $conditions[] = $qb->expr()->and(...$subConditions);
                 }
 
-                if ($conditions) {
+                if (\count($conditions) > 0) {
                     $qb->andWhere($qb->expr()->or(...$conditions));
 
                     foreach ($parameters as $param => $value) {

src/adapter/etl-adapter-doctrine/src/Flow/ETL/Adapter/Doctrine/DbalKeySetExtractor.php

@@ -0,0 +1,103 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Adapter\Doctrine;
+
+use Doctrine\DBAL\{Connection, DriverManager, TransactionIsolationLevel};
+use Flow\ETL\Exception\InvalidArgumentException;
+use Flow\ETL\{FlowContext, Loader, Rows};
+
+final class TransactionalDbalLoader implements Loader
+{
+    private ?Connection $connection = null;
+
+    private TransactionIsolationLevel|int|null $isolationLevel = null;
+
+    /**
+     * @var array<DbalLoader>
+     */
+    private readonly array $loaders;
+
+    /**
+     * @param array<string, mixed> $connectionParams
+     * @param DbalLoader ...$loaders
+     */
+    public function __construct(
+        private readonly array $connectionParams,
+        DbalLoader ...$loaders,
+    ) {
+        if (\count($loaders) === 0) {
+            throw new InvalidArgumentException('At least one loader must be provided');
+        }
+
+        $this->loaders = $loaders;
+    }
+
+    /**
+     * Since Connection::getParams() is marked as an internal method, please
+     * use this constructor with caution.
+     */
+    public static function fromConnection(
+        Connection $connection,
+        DbalLoader ...$loaders,
+    ) : self {
+        $loader = new self($connection->getParams(), ...$loaders);
+        $loader->connection = $connection;
+
+        return $loader;
+    }
+
+    public function load(Rows $rows, FlowContext $context) : void
+    {
+        $this->executeInTransaction($this->connection(), $rows, $context);
+    }
+
+    public function withIsolationLevel(TransactionIsolationLevel|int $level) : self
+    {
+        $this->isolationLevel = $level;
+
+        return $this;
+    }
+
+    private function connection() : Connection
+    {
+        if ($this->connection === null) {
+            /** @phpstan-ignore-next-line */
+            $this->connection = DriverManager::getConnection($this->connectionParams);
+        }
+
+        return $this->connection;
+    }
+
+    private function executeInTransaction(Connection $connection, Rows $rows, FlowContext $context) : void
+    {
+        $previousIsolationLevel = null;
+
+        if ($this->isolationLevel !== null) {
+            $previousIsolationLevel = $connection->getTransactionIsolation();
+            /** @phpstan-ignore-next-line */
+            $connection->setTransactionIsolation($this->isolationLevel);
+        }
+
+        try {
+            $connection->beginTransaction();
+
+            try {
+                foreach ($this->loaders as $loader) {
+                    $loader->load($rows, $context);
+                }
+
+                $connection->commit();
+            } catch (\Throwable $e) {
+                $connection->rollBack();
+
+                throw $e;
+            }
+        } finally {
+            if ($previousIsolationLevel !== null) {
+                $connection->setTransactionIsolation($previousIsolationLevel);
+            }
+        }
+    }
+}

src/adapter/etl-adapter-doctrine/src/Flow/ETL/Adapter/Doctrine/TransactionalDbalLoader.php

@@ -24,6 +24,7 @@
     Attribute\DocumentationExample,
     Attribute\Module,
     Attribute\Type as DSLType,
+    Loader,
     Schema};
 use Flow\ETL\Exception\InvalidArgumentException;
 
@@ -325,6 +326,26 @@ function postgresql_update_options(
     );
 }
 
+/**
+ * Execute multiple loaders within a database transaction.
+ * Each batch of rows will be processed in its own transaction.
+ * If any loader fails, the entire batch will be rolled back.
+ *
+ * @param array<string, mixed>|Connection $connection
+ * @param DbalLoader ...$loaders - DBAL loaders to execute within the transaction
+ *
+ * @throws InvalidArgumentException
+ */
+#[DocumentationDSL(module: Module::DOCTRINE, type: DSLType::LOADER)]
+function to_dbal_transaction(
+    array|Connection $connection,
+    DbalLoader ...$loaders,
+) : TransactionalDbalLoader {
+    return \is_array($connection)
+        ? new TransactionalDbalLoader($connection, ...$loaders)
+        : TransactionalDbalLoader::fromConnection($connection, ...$loaders);
+}
+
 #[DocumentationDSL(module: Module::DOCTRINE, type: DSLType::HELPER)]
 function pagination_key_asc(string $column, string|int|ParameterType|Type $type = ParameterType::STRING) : Key
 {