feat: import export bundle

Author: erdnaxelaweb

components/ImportExportBundle/bin/mdb-to-sqlite.bash

@@ -0,0 +1,17 @@
+#!/usr/bin/env bash
+
+mdb_path=$1
+sqlite_path=$2
+id=$BASHPID
+tmp_dir="/tmp/mdb-convert/${id}"
+
+test -f $sqlite_path && rm $sqlite_path
+
+echo "Starting conversion in ${tmp_dir}"
+mkdir -p $tmp_dir
+mdb-schema $mdb_path sqlite > ${tmp_dir}/schema.sql
+mkdir -p ${tmp_dir}/sql
+for i in $( mdb-tables $mdb_path ); do mdb-export -D "%Y-%m-%d %H:%M:%S" -H -I sqlite $mdb_path $i > ${tmp_dir}/sql/$i.sql; done
+cat ${tmp_dir}/schema.sql | sqlite3 $sqlite_path
+for f in ${tmp_dir}/sql/* ; do (echo 'BEGIN;'; cat $f; echo 'COMMIT;') | sqlite3 $sqlite_path; done
+rm -rf $tmp_dir

components/ImportExportBundle/composer.json

@@ -34,5 +34,8 @@
     "symfony/messenger": "Used to run job asynchronously"
   },
   "extra": {
-  }
+  },
+  "bin": [
+    "bin/mdb-to-sqlite.bash"
+  ]
 }

components/ImportExportBundle/src/bundle/Command/ExecuteWorkflowCommand.php

@@ -43,7 +43,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int
     {
         $workflowIdentifier = $input->getArgument('identifier');
         $workflow = $this->workflowRegistry->getWorkflow($workflowIdentifier);
-        $baseConfiguration = $workflow::getDefaultConfig();
+        $baseConfiguration = $workflow->getDefaultConfig();
         if (!$baseConfiguration->isAvailable(WorkflowConfiguration::AVAILABILITY_CLI)) {
             throw new InvalidArgumentException(sprintf('Workflow %s is not available', $workflowIdentifier));
         }

components/ImportExportBundle/src/bundle/Resources/config/forms.yaml

@@ -42,3 +42,8 @@ services:
             $filepathResolver: '@AlmaviaCX\Bundle\IbexaImportExport\Resolver\FilepathResolver'
         tags:
             - { name: form.type, alias: StreamWriterOptionsFormType }
+
+    AlmaviaCX\Bundle\IbexaImportExport\Reader\Mdb\MdbReaderOptionsFormType:
+        parent: AlmaviaCX\Bundle\IbexaImportExport\Reader\File\FileReaderOptionsFormType
+        tags:
+            - { name: form.type, alias: MdbReaderOptionsFormType }

components/ImportExportBundle/src/bundle/Resources/config/workflow/component/reader.yaml

@@ -28,3 +28,10 @@ services:
         parent: AlmaviaCX\Bundle\IbexaImportExport\Reader\File\AbstractFileReader
         tags:
             - { name: almaviacx.import_export.component, alias: reader.xml}
+
+    AlmaviaCX\Bundle\IbexaImportExport\Reader\Mdb\MdbReader:
+        parent: AlmaviaCX\Bundle\IbexaImportExport\Reader\File\AbstractFileReader
+        arguments:
+            $converterPath: '%kernel.project_dir%/bin/mdb-to-sqlite.bash'
+        tags:
+            - { name: almaviacx.import_export.component, alias: reader.mdb}

components/ImportExportBundle/src/lib/Component/AbstractComponent.php

@@ -49,6 +49,14 @@ public function clean(): void
     {
     }
 
+    public function prepare(): void
+    {
+    }
+
+    public function finish(): void
+    {
+    }
+
     public function setLogger(WorkflowLoggerInterface $logger): void
     {
         $this->logger = $logger;

components/ImportExportBundle/src/lib/Component/ComponentInterface.php

@@ -27,4 +27,8 @@ public function getOptions(): ComponentOptions;
     public function clean(): void;
 
     public function setLogger(WorkflowLoggerInterface $logger): void;
+
+    public function prepare(): void;
+
+    public function finish(): void;
 }

components/ImportExportBundle/src/lib/Item/Iterator/DoctrineSeekableItemIterator.php

@@ -0,0 +1,106 @@
+<?php
+
+declare(strict_types=1);
+
+namespace AlmaviaCX\Bundle\IbexaImportExport\Item\Iterator;
+
+use AlmaviaCX\Bundle\IbexaImportExport\Reader\ReaderIteratorInterface;
+use ArrayIterator;
+use Doctrine\DBAL\Connection;
+use Iterator;
+use SeekableIterator;
+
+class DoctrineSeekableItemIterator implements ReaderIteratorInterface, SeekableIterator
+{
+    public const DEFAULT_BATCH_SIZE = 25;
+    protected Connection $connection;
+    protected string $queryString;
+    protected string $countQueryString;
+    protected int $batchSize = self::DEFAULT_BATCH_SIZE;
+    private ?Iterator $innerIterator;
+    private int $position = 0;
+
+    public function __construct(
+        Connection $connection,
+        string $queryString,
+        string $countQueryString,
+        int $batchSize = self::DEFAULT_BATCH_SIZE
+    ) {
+        $this->batchSize = $batchSize;
+        $this->countQueryString = $countQueryString;
+        $this->queryString = $queryString;
+        $this->connection = $connection;
+    }
+
+    private function fetch(): Iterator
+    {
+        $queryString = sprintf('%s LIMIT %d OFFSET %d', $this->queryString, $this->batchSize, $this->position);
+
+        return new ArrayIterator($this->connection->executeQuery($queryString)->fetchAllAssociative());
+    }
+
+    private function initialize(): void
+    {
+        $this->position = 0;
+        $this->innerIterator = $this->fetch();
+    }
+
+    private function isInitialized(): bool
+    {
+        return isset($this->innerIterator);
+    }
+
+    /**
+     * @return mixed
+     */
+    public function current()
+    {
+        if (!$this->isInitialized()) {
+            $this->initialize();
+        }
+
+        return $this->innerIterator->current();
+    }
+
+    public function next(): void
+    {
+        if (!$this->isInitialized()) {
+            $this->initialize();
+        }
+        ++$this->position;
+        $this->innerIterator->next();
+        if (!$this->innerIterator->valid() && ($this->position % $this->batchSize) === 0) {
+            $this->innerIterator = $this->fetch();
+        }
+    }
+
+    public function key(): mixed
+    {
+        return $this->position;
+    }
+
+    public function valid(): bool
+    {
+        if (!$this->isInitialized()) {
+            $this->initialize();
+        }
+
+        return $this->innerIterator->valid();
+    }
+
+    public function rewind(): void
+    {
+        $this->initialize();
+    }
+
+    public function count(): int
+    {
+        return $this->connection->executeQuery($this->countQueryString)->fetchOne();
+    }
+
+    public function seek($offset): void
+    {
+        $this->position = $offset;
+        $this->innerIterator = $this->fetch();
+    }
+}

components/ImportExportBundle/src/lib/Item/Iterator/ItemIterator.php

@@ -10,12 +10,12 @@
 class ItemIterator extends AbstractReaderIterator
 {
     protected Iterator $innerIterator;
-    protected IteratorItemTransformerInterface $itemTransformer;
+    protected ?IteratorItemTransformerInterface $itemTransformer = null;
 
     public function __construct(
         int $totalCount,
         Iterator $innerIterator,
-        IteratorItemTransformerInterface $itemTransformer
+        ?IteratorItemTransformerInterface $itemTransformer = null
     ) {
         $this->itemTransformer = $itemTransformer;
         $this->innerIterator = $innerIterator;
@@ -24,7 +24,12 @@ public function __construct(
 
     public function current()
     {
-        return ($this->itemTransformer)($this->innerIterator->current());
+        $item = $this->innerIterator->current();
+        if ($this->itemTransformer instanceof IteratorItemTransformerInterface) {
+            return ($this->itemTransformer)($item);
+        }
+
+        return $item;
     }
 
     public function next(): void

components/ImportExportBundle/src/lib/Job/Job.php

@@ -94,7 +94,7 @@ class Job
      *
      * @var Collection<JobRecord>
      */
-    protected Collection $records;
+    protected ?Collection $records = null;
 
     /**
      * @ORM\Column(type="text", nullable=true)