add depentency to doctrine/doctrine-bundle (#29 )

2026-03-24 06:42:23 +01:00 · 2019-11-08 14:58:46 +01:00
27 changed files with 251 additions and 936 deletions
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,31 +0,0 @@
-# Version 2.0.2
-
-* Fixed the connection proxy class created by the factory
-
-# Version 2.0.1
-
-* Fixed next execution time not increasing for scheduled dataflows
-
-# Version 2.0.0
-
-* Add Doctrine DBAL multi-connection support
-* Add configuration to define the default Doctrine DBAL connection
-* Remove Doctrine ORM
-* Rewrite repositories
-
-# Version 1.0.1
-
-* Fix lost dependency
-* Fix schedule removing
-
-# Version 1.0.0
-
-Initial version
-
-* Define and configure a Dataflow
-* Run the Job scheduled
-* Run one Dataflow from the command line
-* Define the schedule for a Dataflow from the command line
-* Enable/Disable a scheduled Dataflow from the command line
-* Display the list of scheduled Dataflow from the command line
-* Display the result for the last Job for a Dataflow from the command line
--- a/README.md
+++ b/README.md
@@ -30,7 +30,6 @@ As the following schema shows, you can define more than one dataflow:
 * Enable/Disable a scheduled Dataflow from the command line
 * Display the list of scheduled Dataflow from the command line
 * Display the result for the last Job for a Dataflow from the command line
-* Work with multiple Doctrine DBAL connections


 ## Installation
@@ -40,7 +39,7 @@ As the following schema shows, you can define more than one dataflow:
 To install this bundle, run this command :

 ```shell script
-$ composer require code-rhapsodie/dataflow-bundle
+$ composer require code-rhapsodie/dataflow
 ```

 #### Suggest
@@ -99,32 +98,26 @@ public function registerBundles()

 ### Update the database

-This bundle uses Doctrine DBAL to store Dataflow schedule into the database table (`cr_dataflow_scheduled`)
+This bundle uses Doctrine ORM for drive the database table for store Dataflow schedule (`cr_dataflow_scheduled`)
 and jobs (`cr_dataflow_job`).

-If you use [Doctrine Migration Bundle](https://symfony.com/doc/master/bundles/DoctrineMigrationsBundle/index.html) or [Phinx](https://phinx.org/) 
-or [Kaliop Migration Bundle](https://github.com/kaliop-uk/ezmigrationbundle) or whatever,
+#### Doctrine migration
+
+Execute the command to generate the migration for your database:
+
+```shell script
+$ bin/console doctrine:migration:diff
+```
+
+#### Other migration tools
+
+If you use [Phinx](https://phinx.org/) or [Kaliop Migration Bundle](https://github.com/kaliop-uk/ezmigrationbundle) or whatever,
 you can add a new migration with the generated SQL query from this command:

 ```shell script
-$ bin/console code-rhapsodie:dataflow:dump-schema
+$ bin/console doctrine:schema:update --dump-sql
 ```

-If you have already the tables, you can add a new migration with the generated update SQL query from this command:
-
-```shell script
-$ bin/console code-rhapsodie:dataflow:dump-schema --update
-```
-
-## Configuration
-
-By default, the Doctrine DBAL connection used is `default`. You can configure the default connection.
-Add this configuration into your Symfony configuration:
-
-```yaml
-code_rhapsodie_dataflow:
-  dbal_default_connection: test #Name of the default connection used by Dataflow bundle
-```

 ## Define a dataflow type

@@ -167,11 +160,11 @@ class MyFirstDataflowType extends AbstractDataflowType

    protected function buildDataflow(DataflowBuilder $builder, array $options): void
    {
-        $this->myWriter->setDestinationFilePath($options['to-file']);
+        
+        $this->myReader->setFilename($options['fileName']);

-        $builder
-            ->setReader($this->myReader->read($options['from-file']))
-            ->addStep(function ($data) use ($options) {
+        $builder->setReader($this->myReader)
+            ->addStep(function($data) use ($options) {
                // TODO : Write your code here...
                return $data;
            })
@@ -181,8 +174,11 @@ class MyFirstDataflowType extends AbstractDataflowType

    protected function configureOptions(OptionsResolver $optionsResolver): void
    {
-        $optionsResolver->setDefaults(['to-file' => '/tmp/dataflow.csv', 'from-file' => null]);
-        $optionsResolver->setRequired('from-file');
+        $optionsResolver->setDefaults([
+            'my_option' => 'my_default_value',
+            'fileName'  => null,
+        ]);
+        $optionsResolver->setRequired('fileName');
    }

    public function getLabel(): string
@@ -226,8 +222,11 @@ class MyFirstDataflowType extends AbstractDataflowType
    // ...
    protected function configureOptions(OptionsResolver $optionsResolver): void
    {
-        $optionsResolver->setDefaults(['to-file' => '/tmp/dataflow.csv', 'from-file' => null]);
-        $optionsResolver->setRequired('from-file');
+        $optionsResolver->setDefaults([
+            'my_option' => 'my_default_value',
+            'fileName'  => null,
+        ]);
+        $optionsResolver->setRequired('fileName');
    }

 }
@@ -275,18 +274,27 @@ namespace CodeRhapsodie\DataflowExemple\Reader;

 class FileReader
 {
-    public function read(string $filename): iterable
+    private $filename;
+
+    /**
+     * Set the filename option needed by the Reader.
+     */
+    public function setFilename(string $filename) {
+        $this->filename = $filename;
+    }
+
+    public function __invoke(): iterable
    {
-        if (!$filename) {
+        if (!$this->filename) {
            throw new \Exception("The file name is not defined. Define it with 'setFilename' method");
        }

-        if (!$fh = fopen($filename, 'r')) {
-            throw new \Exception("Unable to open file '".$filename."' for read.");
+        if (!$fh = fopen($this->filename, 'r')) {
+            throw new \Exception("Unable to open file '".$this->filename."' for read.");
        }

-        while (false !== ($read = fgets($fh))) {
-            yield explode('|', trim($read));
+        while (false === ($read = fread($fh, 1024))) {
+            yield explode("|", $read);
        }
    }
 }
@@ -312,16 +320,14 @@ A *Step* can be any callable, taking the element as its argument, and returning
 A few examples:

 ```php
-<?php
-//[...]
-$builder->addStep(function ($item) {
+$builder->addStep(function($item) {
    // Titles are changed to all caps before export
    $item['title'] = strtoupper($item['title']);

    return $item;
 });

-$builder->addStep(function ($item) {
+$builder->addStep(function($item) {
    // Private items are not exported
    if ($item['private']) {
        return false;
@@ -329,7 +335,6 @@ $builder->addStep(function ($item) {

    return $item;
 });
-//[...]
 ```

 ### Writers
@@ -357,20 +362,11 @@ class FileWriter implements WriterInterface
 {
    private $fh;

-    /** @var string */
-    private $path;
-
-    public function setDestinationFilePath(string $path) {
-        $this->path = $path;
-    }
-
    public function prepare()
    {
-        if (null === $this->path) {
-            throw new \Exception('Define the destination file name before use');
-        }
-        if (!$this->fh = fopen($this->path, 'w')) {
-            throw new \Exception('Unable to open in write mode the output file.');
+
+        if (!$this->fh = fopen('/path/to/file', 'w')) {
+            throw new \Exception("Unable to open in write mode the output file.");
        }
    }

@@ -412,28 +408,6 @@ Several commands are provided to manage schedules and run jobs.

 `code-rhapsodie:dataflow:execute` Let you execute one dataflow job.

-`code-rhapsodie:dataflow:dump-schema` Generates schema create / update SQL queries
-
-### Work with many databases
-
-All commands have a `--connection` option to define what Doctrine DBAL connection to use during execution.
-
-Example:
-
-This command uses the `default` DBAL connection to generate all schema update queries.
-
-```shell script
-$ bin/console code-rhapsodie:dataflow:dump-schema --update --connection=default
-```
-
-To execute all pending job for a specific connection use:
-
-```shell script
-# Run for dataflow DBAL connection
-$ bin/console code-rhapsodie:dataflow:run-pending --connection=dataflow
-# Run for default DBAL connection
-$ bin/console code-rhapsodie:dataflow:run-pending --connection=default
-```

 # Issues and feature requests

--- a/Tests/Manager/ScheduledDataflowManagerTest.php
+++ b/Tests/Manager/ScheduledDataflowManagerTest.php
@@ -9,7 +9,6 @@ use CodeRhapsodie\DataflowBundle\Exceptions\UnknownDataflowTypeException;
 use CodeRhapsodie\DataflowBundle\Manager\ScheduledDataflowManager;
 use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
 use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
-use Doctrine\DBAL\Driver\Connection;
 use Doctrine\ORM\EntityManagerInterface;
 use PHPUnit\Framework\MockObject\MockObject;
 use PHPUnit\Framework\TestCase;
@@ -19,8 +18,8 @@ class ScheduledDataflowManagerTest extends TestCase
    /** @var ScheduledDataflowManager */
    private $manager;

-    /** @var Connection|MockObject */
-    private $connection;
+    /** @var EntityManagerInterface|MockObject */
+    private $em;

    /** @var ScheduledDataflowRepository|MockObject */
    private $scheduledDataflowRepository;
@@ -30,18 +29,17 @@ class ScheduledDataflowManagerTest extends TestCase

    protected function setUp(): void
    {
-        $this->connection = $this->createMock(Connection::class);
+        $this->em = $this->createMock(EntityManagerInterface::class);
        $this->scheduledDataflowRepository = $this->createMock(ScheduledDataflowRepository::class);
        $this->jobRepository = $this->createMock(JobRepository::class);

-        $this->manager = new ScheduledDataflowManager($this->connection, $this->scheduledDataflowRepository, $this->jobRepository);
+        $this->manager = new ScheduledDataflowManager($this->em, $this->scheduledDataflowRepository, $this->jobRepository);
    }

    public function testCreateJobsFromScheduledDataflows()
    {
        $scheduled1 = new ScheduledDataflow();
        $scheduled2 = (new ScheduledDataflow())
-            ->setId(-1)
            ->setDataflowType($type = 'testType')
            ->setOptions($options = ['opt' => 'val'])
            ->setNext($next = new \DateTime())
@@ -62,13 +60,9 @@ class ScheduledDataflowManagerTest extends TestCase
            ->willReturnOnConsecutiveCalls(new Job(), null)
        ;

-        $this->connection
+        $this->em
            ->expects($this->once())
-            ->method('beginTransaction')
-        ;
-        $this->jobRepository
-            ->expects($this->once())
-            ->method('save')
+            ->method('persist')
            ->with(
                $this->callback(function (Job $job) use ($type, $options, $next, $label, $scheduled2) {
                    return (
@@ -77,21 +71,15 @@ class ScheduledDataflowManagerTest extends TestCase
                        && $job->getOptions() === $options
                        && $job->getRequestedDate() == $next
                        && $job->getLabel() === $label
-                        && $job->getScheduledDataflowId() === $scheduled2->getId()
+                        && $job->getScheduledDataflow() === $scheduled2
                    );
                })
            )
        ;

-        $this->scheduledDataflowRepository
+        $this->em
            ->expects($this->once())
-            ->method('save')
-            ->with($scheduled2)
-        ;
-
-        $this->connection
-            ->expects($this->once())
-            ->method('commit')
+            ->method('flush')
        ;

        $this->manager->createJobsFromScheduledDataflows();
--- a/Tests/Runner/PendingDataflowRunnerTest.php
+++ b/Tests/Runner/PendingDataflowRunnerTest.php
@@ -20,6 +20,9 @@ class PendingDataflowRunnerTest extends TestCase
    /** @var PendingDataflowRunner */
    private $runner;

+    /** @var EntityManagerInterface|MockObject */
+    private $em;
+
    /** @var JobRepository|MockObject */
    private $repository;

@@ -31,11 +34,12 @@ class PendingDataflowRunnerTest extends TestCase

    protected function setUp(): void
    {
+        $this->em = $this->createMock(EntityManagerInterface::class);
        $this->repository = $this->createMock(JobRepository::class);
        $this->registry = $this->createMock(DataflowTypeRegistryInterface::class);
        $this->dispatcher = $this->createMock(EventDispatcherInterface::class);

-        $this->runner = new PendingDataflowRunner($this->repository, $this->registry, $this->dispatcher);
+        $this->runner = new PendingDataflowRunner($this->em, $this->repository, $this->registry, $this->dispatcher);
    }

    public function testRunPendingDataflows()
@@ -118,9 +122,9 @@ class PendingDataflowRunnerTest extends TestCase
            ->willReturn($result2)
        ;

-        $this->repository
+        $this->em
            ->expects($this->exactly(4))
-            ->method('save')
+            ->method('flush')
        ;

        $this->runner->runPendingDataflows();
--- a/UPGRADE.md
+++ b/UPGRADE.md
@@ -1,3 +0,0 @@
-# Upgrade from v1.x to v2.0
-
-[BC] `JobRepository` and `ScheduledDataflowRepository` are no longer a Doctrine ORM repository.
--- a/composer.json
+++ b/composer.json
@@ -37,7 +37,7 @@
    },
    "require": {
        "php": "^7.1",
-        "doctrine/dbal": "^2.0",
+        "doctrine/orm": "^2.4.5",
        "seld/signal-handler": "^1.0",
        "symfony/config": "^3.4||^4.0",
        "symfony/console": "^3.4||^4.0",
@@ -62,8 +62,7 @@
    },
    "extra": {
        "branch-alias": {
-            "dev-master": "2.x-dev",
-            "dev-v1.x": "1.x-dev"
+            "dev-master": "1.x-dev"
        }
    }
 }
--- a/src/Command/AddScheduledDataflowCommand.php
+++ b/src/Command/AddScheduledDataflowCommand.php
@@ -13,7 +13,6 @@ use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 use Symfony\Component\Console\Style\SymfonyStyle;
 use Symfony\Component\Validator\Validator\ValidatorInterface;
-use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;

 /**
 * @codeCoverageIgnore
@@ -29,17 +28,13 @@ class AddScheduledDataflowCommand extends Command
    /** @var ValidatorInterface */
    private $validator;

-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(DataflowTypeRegistryInterface $registry, ScheduledDataflowRepository $scheduledDataflowRepository, ValidatorInterface $validator, ConnectionFactory $connectionFactory)
+    public function __construct(DataflowTypeRegistryInterface $registry, ScheduledDataflowRepository $scheduledDataflowRepository, ValidatorInterface $validator)
    {
        parent::__construct();

        $this->registry = $registry;
        $this->scheduledDataflowRepository = $scheduledDataflowRepository;
        $this->validator = $validator;
-        $this->connectionFactory = $connectionFactory;
    }

    /**
@@ -52,12 +47,10 @@ class AddScheduledDataflowCommand extends Command
            ->setHelp('The <info>%command.name%</info> allows you to create a new scheduled dataflow.')
            ->addOption('label', null, InputOption::VALUE_REQUIRED, 'Label of the scheduled dataflow')
            ->addOption('type', null, InputOption::VALUE_REQUIRED, 'Type of the scheduled dataflow (FQCN)')
-            ->addOption('options', null, InputOption::VALUE_OPTIONAL,
-                'Options of the scheduled dataflow (ex: {"option1": "value1", "option2": "value2"})')
+            ->addOption('options', null, InputOption::VALUE_OPTIONAL, 'Options of the scheduled dataflow (ex: {"option1": "value1", "option2": "value2"})')
            ->addOption('frequency', null, InputOption::VALUE_REQUIRED, 'Frequency of the scheduled dataflow')
            ->addOption('first_run', null, InputOption::VALUE_REQUIRED, 'Date for the first run of the scheduled dataflow (Y-m-d H:i:s)')
-            ->addOption('enabled', null, InputOption::VALUE_REQUIRED, 'State of the scheduled dataflow')
-            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
+            ->addOption('enabled', null, InputOption::VALUE_REQUIRED, 'State of the scheduled dataflow');
    }

    /**
@@ -65,9 +58,6 @@ class AddScheduledDataflowCommand extends Command
     */
    protected function execute(InputInterface $input, OutputInterface $output)
    {
-        if (null !== $input->getOption('connection')) {
-            $this->connectionFactory->setConnectionName($input->getOption('connection'));
-        }
        $choices = [];
        $typeMapping = [];
        foreach ($this->registry->listDataflowTypes() as $fqcn => $dataflowType) {
@@ -87,13 +77,11 @@ class AddScheduledDataflowCommand extends Command
        }
        $options = $input->getOption('options');
        if (!$options) {
-            $options = $io->ask('What are the launch options for the scheduled dataflow? (ex: {"option1": "value1", "option2": "value2"})',
-                json_encode([]));
+            $options = $io->ask('What are the launch options for the scheduled dataflow? (ex: {"option1": "value1", "option2": "value2"})', json_encode([]));
        }
        $frequency = $input->getOption('frequency');
        if (!$frequency) {
-            $frequency = $io->choice('What is the frequency for the scheduled dataflow?',
-                ScheduledDataflow::AVAILABLE_FREQUENCIES);
+            $frequency = $io->choice('What is the frequency for the scheduled dataflow?', ScheduledDataflow::AVAILABLE_FREQUENCIES);
        }
        $firstRun = $input->getOption('first_run');
        if (!$firstRun) {
@@ -104,27 +92,44 @@ class AddScheduledDataflowCommand extends Command
            $enabled = $io->confirm('Enable the scheduled dataflow?');
        }

-        $newScheduledDataflow = ScheduledDataflow::createFromArray([
-            'id' => null,
-            'label' => $label,
-            'dataflow_type' => $type,
-            'options' => json_decode($options, true),
-            'frequency' => $frequency,
-            'next' => new \DateTimeImmutable($firstRun),
-            'enabled' => $enabled,
-        ]);
+        try {
+            $newScheduledDataflow = $this->createEntityFromArray([
+                'label' => $label,
+                'type' => $type,
+                'options' => $options,
+                'frequency' => $frequency,
+                'first_run' => $firstRun,
+                'enabled' => $enabled,
+            ]);

-        $errors = $this->validator->validate($newScheduledDataflow);
-        if (count($errors) > 0) {
-            $io->error((string) $errors);
+            $errors = $this->validator->validate($newScheduledDataflow);
+            if (count($errors) > 0) {
+                $io->error((string) $errors);

-            return 2;
+                return 2;
+            }
+
+            $this->scheduledDataflowRepository->save($newScheduledDataflow);
+            $io->success(sprintf('New scheduled dataflow "%s" (id:%d) was created successfully.', $newScheduledDataflow->getLabel(), $newScheduledDataflow->getId()));
+
+            return 0;
+        } catch (\Exception $e) {
+            $io->error(sprintf('An error occured when creating new scheduled dataflow : "%s".', $e->getMessage()));
+
+            return 1;
        }
+    }

-        $this->scheduledDataflowRepository->save($newScheduledDataflow);
-        $io->success(sprintf('New scheduled dataflow "%s" (id:%d) was created successfully.',
-            $newScheduledDataflow->getLabel(), $newScheduledDataflow->getId()));
+    private function createEntityFromArray(array $input): ScheduledDataflow
+    {
+        $scheduledDataflow = new ScheduledDataflow();
+        $scheduledDataflow->setLabel($input['label']);
+        $scheduledDataflow->setDataflowType($input['type']);
+        $scheduledDataflow->setOptions(json_decode($input['options'], true));
+        $scheduledDataflow->setFrequency($input['frequency']);
+        $scheduledDataflow->setNext(new \DateTimeImmutable($input['first_run']));
+        $scheduledDataflow->setEnabled($input['enabled']);

-        return 0;
+        return $scheduledDataflow;
    }
 }
--- a/src/Command/ChangeScheduleStatusCommand.php
+++ b/src/Command/ChangeScheduleStatusCommand.php
@@ -12,7 +12,6 @@ use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 use Symfony\Component\Console\Style\SymfonyStyle;
-use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;

 /**
 * @codeCoverageIgnore
@@ -24,15 +23,11 @@ class ChangeScheduleStatusCommand extends Command
    /** @var ScheduledDataflowRepository */
    private $scheduledDataflowRepository;

-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(ScheduledDataflowRepository $scheduledDataflowRepository, ConnectionFactory $connectionFactory)
+    public function __construct(ScheduledDataflowRepository $scheduledDataflowRepository)
    {
        parent::__construct();

        $this->scheduledDataflowRepository = $scheduledDataflowRepository;
-        $this->connectionFactory = $connectionFactory;
    }

    /**
@@ -45,8 +40,7 @@ class ChangeScheduleStatusCommand extends Command
            ->setHelp('The <info>%command.name%</info> command able you to change schedule status.')
            ->addArgument('schedule-id', InputArgument::REQUIRED, 'Id of the schedule')
            ->addOption('enable', null, InputOption::VALUE_NONE, 'Enable the schedule')
-            ->addOption('disable', null, InputOption::VALUE_NONE, 'Disable the schedule')
-            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
+            ->addOption('disable', null, InputOption::VALUE_NONE, 'Disable the schedule');
    }

    /**
@@ -54,12 +48,9 @@ class ChangeScheduleStatusCommand extends Command
     */
    protected function execute(InputInterface $input, OutputInterface $output)
    {
-        if (null !== $input->getOption('connection')) {
-            $this->connectionFactory->setConnectionName($input->getOption('connection'));
-        }
        $io = new SymfonyStyle($input, $output);
        /** @var ScheduledDataflow|null $schedule */
-        $schedule = $this->scheduledDataflowRepository->find((int) $input->getArgument('schedule-id'));
+        $schedule = $this->scheduledDataflowRepository->find($input->getArgument('schedule-id'));

        if (!$schedule) {
            $io->error(sprintf('Cannot find scheduled dataflow with id "%d".', $input->getArgument('schedule-id')));
--- a/src/Command/ExecuteDataflowCommand.php
+++ b/src/Command/ExecuteDataflowCommand.php
@@ -8,9 +8,7 @@ use CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
-use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
-use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;

 /**
 * Runs one dataflow.
@@ -24,15 +22,11 @@ class ExecuteDataflowCommand extends Command
    /** @var DataflowTypeRegistryInterface */
    private $registry;

-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(DataflowTypeRegistryInterface $registry, ConnectionFactory $connectionFactory)
+    public function __construct(DataflowTypeRegistryInterface $registry)
    {
        parent::__construct();

        $this->registry = $registry;
-        $this->connectionFactory = $connectionFactory;
    }

    /**
@@ -50,7 +44,7 @@ EOF
            )
            ->addArgument('fqcn', InputArgument::REQUIRED, 'FQCN or alias of the dataflow type')
            ->addArgument('options', InputArgument::OPTIONAL, 'Options for the dataflow type as a json string', '[]')
-            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
+        ;
    }

    /**
@@ -58,9 +52,6 @@ EOF
     */
    protected function execute(InputInterface $input, OutputInterface $output)
    {
-        if (null !== $input->getOption('connection')) {
-            $this->connectionFactory->setConnectionName($input->getOption('connection'));
-        }
        $fqcnOrAlias = $input->getArgument('fqcn');
        $options = json_decode($input->getArgument('options'), true);

--- a/src/Command/JobShowCommand.php
+++ b/src/Command/JobShowCommand.php
@@ -11,7 +11,6 @@ use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 use Symfony\Component\Console\Style\SymfonyStyle;
-use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;

 /**
 * @codeCoverageIgnore
@@ -29,15 +28,11 @@ class JobShowCommand extends Command
    /** @var JobRepository */
    private $jobRepository;

-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(JobRepository $jobRepository, ConnectionFactory $connectionFactory)
+    public function __construct(JobRepository $jobRepository)
    {
        parent::__construct();

        $this->jobRepository = $jobRepository;
-        $this->connectionFactory = $connectionFactory;
    }

    /**
@@ -50,8 +45,7 @@ class JobShowCommand extends Command
            ->setHelp('The <info>%command.name%</info> display job details for schedule or specific job.')
            ->addOption('job-id', null, InputOption::VALUE_REQUIRED, 'Id of the job to get details')
            ->addOption('schedule-id', null, InputOption::VALUE_REQUIRED, 'Id of schedule for last execution details')
-            ->addOption('details', null, InputOption::VALUE_NONE, 'Display full details')
-            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
+            ->addOption('details', null, InputOption::VALUE_NONE, 'Display full details');
    }

    /**
@@ -59,10 +53,6 @@ class JobShowCommand extends Command
     */
    protected function execute(InputInterface $input, OutputInterface $output)
    {
-        if (null !== $input->getOption('connection')) {
-            $this->connectionFactory->setConnectionName($input->getOption('connection'));
-        }
-
        $io = new SymfonyStyle($input, $output);

        $jobId = (int) $input->getOption('job-id');
--- a/src/Command/RunPendingDataflowsCommand.php
+++ b/src/Command/RunPendingDataflowsCommand.php
@@ -9,9 +9,7 @@ use CodeRhapsodie\DataflowBundle\Runner\PendingDataflowRunnerInterface;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Command\LockableTrait;
 use Symfony\Component\Console\Input\InputInterface;
-use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
-use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;

 /**
 * Runs dataflows according to user-defined schedule.
@@ -30,16 +28,12 @@ class RunPendingDataflowsCommand extends Command
    /** @var PendingDataflowRunnerInterface */
    private $runner;

-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(ScheduledDataflowManagerInterface $manager, PendingDataflowRunnerInterface $runner, ConnectionFactory $connectionFactory)
+    public function __construct(ScheduledDataflowManagerInterface $manager, PendingDataflowRunnerInterface $runner)
    {
        parent::__construct();

        $this->manager = $manager;
        $this->runner = $runner;
-        $this->connectionFactory = $connectionFactory;
    }

    /**
@@ -53,7 +47,7 @@ class RunPendingDataflowsCommand extends Command
 The <info>%command.name%</info> command runs dataflows according to the schedule defined in the UI by the user.
 EOF
            )
-            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
+        ;
    }

    /**
@@ -67,10 +61,6 @@ EOF
            return 0;
        }

-        if (null !== $input->getOption('connection')) {
-            $this->connectionFactory->setConnectionName($input->getOption('connection'));
-        }
-
        $this->manager->createJobsFromScheduledDataflows();
        $this->runner->runPendingDataflows();

--- a/src/Command/ScheduleListCommand.php
+++ b/src/Command/ScheduleListCommand.php
@@ -7,10 +7,8 @@ namespace CodeRhapsodie\DataflowBundle\Command;
 use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputInterface;
-use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 use Symfony\Component\Console\Style\SymfonyStyle;
-use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;

 /**
 * @codeCoverageIgnore
@@ -22,15 +20,11 @@ class ScheduleListCommand extends Command
    /** @var ScheduledDataflowRepository */
    private $scheduledDataflowRepository;

-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(ScheduledDataflowRepository $scheduledDataflowRepository, ConnectionFactory $connectionFactory)
+    public function __construct(ScheduledDataflowRepository $scheduledDataflowRepository)
    {
        parent::__construct();

        $this->scheduledDataflowRepository = $scheduledDataflowRepository;
-        $this->connectionFactory = $connectionFactory;
    }

    /**
@@ -40,8 +34,7 @@ class ScheduleListCommand extends Command
    {
        $this
            ->setDescription('List scheduled dataflows')
-            ->setHelp('The <info>%command.name%</info> lists all scheduled dataflows.')
-            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use');
+            ->setHelp('The <info>%command.name%</info> lists all scheduled dataflows.');
    }

    /**
@@ -49,9 +42,6 @@ class ScheduleListCommand extends Command
     */
    protected function execute(InputInterface $input, OutputInterface $output)
    {
-        if (null !== $input->getOption('connection')) {
-            $this->connectionFactory->setConnectionName($input->getOption('connection'));
-        }
        $io = new SymfonyStyle($input, $output);
        $display = [];
        $schedules = $this->scheduledDataflowRepository->listAllOrderedByLabel();
@@ -61,7 +51,7 @@ class ScheduleListCommand extends Command
                $schedule['label'],
                $schedule['enabled'] ? 'yes' : 'no',
                $schedule['startTime'] ? (new \DateTime($schedule['startTime']))->format('Y-m-d H:i:s') : '-',
-                $schedule['next'] ? (new \DateTime($schedule['next']))->format('Y-m-d H:i:s') : '-',
+                $schedule['next'] ? $schedule['next']->format('Y-m-d H:i:s') : '-',
            ];
        }

--- a/src/Command/SchemaCommand.php
+++ b/src/Command/SchemaCommand.php
@@ -1,99 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace CodeRhapsodie\DataflowBundle\Command;
-
-use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
-use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
-use CodeRhapsodie\DataflowBundle\SchemaProvider\DataflowSchemaProvider;
-use Doctrine\DBAL\Schema\Schema;
-use Doctrine\DBAL\Schema\Table;
-use Symfony\Component\Console\Command\Command;
-use Symfony\Component\Console\Input\InputInterface;
-use Symfony\Component\Console\Input\InputOption;
-use Symfony\Component\Console\Output\OutputInterface;
-use Symfony\Component\Console\Style\SymfonyStyle;
-use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
-
-/**
- * @codeCoverageIgnore
- */
-class SchemaCommand extends Command
-{
-    protected static $defaultName = 'code-rhapsodie:dataflow:dump-schema';
-
-    /** @var ConnectionFactory */
-    private $connectionFactory;
-
-    public function __construct(ConnectionFactory $connectionFactory)
-    {
-        parent::__construct();
-
-        $this->connectionFactory = $connectionFactory;
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    protected function configure()
-    {
-        $this
-            ->setDescription('Generates schema create / update SQL queries')
-            ->setHelp('The <info>%command.name%</info> help you to generate SQL Query to create or update your database schema for this bundle')
-            ->addOption('update', null, InputOption::VALUE_NONE, 'Dump only the update SQL queries.')
-            ->addOption('connection', null, InputOption::VALUE_REQUIRED, 'Define the DBAL connection to use')
-        ;
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    protected function execute(InputInterface $input, OutputInterface $output)
-    {
-        if (null !== $input->getOption('connection')) {
-            $this->connectionFactory->setConnectionName($input->getOption('connection'));
-        }
-
-        $connection = $this->connectionFactory->getConnection();
-
-        $schemaProvider = new DataflowSchemaProvider();
-        $schema = $schemaProvider->createSchema();
-
-        $sqls = $schema->toSql($connection->getDatabasePlatform());
-
-        if ($input->getOption('update')) {
-            $sm = $connection->getSchemaManager();
-
-            $tableArray = [JobRepository::TABLE_NAME, ScheduledDataflowRepository::TABLE_NAME];
-            $tables = [];
-            foreach ($sm->listTables() as $table) {
-                /** @var Table $table */
-                if (in_array($table->getName(), $tableArray)) {
-                    $tables[] = $table;
-                }
-            }
-
-            $namespaces = [];
-
-            if ($connection->getDatabasePlatform()->supportsSchemas()) {
-                $namespaces = $sm->listNamespaceNames();
-            }
-
-            $sequences = [];
-
-            if ($connection->getDatabasePlatform()->supportsSequences()) {
-                $sequences = $sm->listSequences();
-            }
-
-            $oldSchema = new Schema($tables, $sequences, $sm->createSchemaConfig(), $namespaces);
-
-            $sqls = $schema->getMigrateFromSql($oldSchema, $connection->getDatabasePlatform());
-        }
-        $io = new SymfonyStyle($input, $output);
-        $io->text('Execute these SQL Queries on your database:');
-        foreach ($sqls as $sql) {
-            $io->text($sql);
-        }
-    }
-}
--- a/src/DataflowType/Dataflow/Dataflow.php
+++ b/src/DataflowType/Dataflow/Dataflow.php
@@ -6,6 +6,7 @@ namespace CodeRhapsodie\DataflowBundle\DataflowType\Dataflow;

 use CodeRhapsodie\DataflowBundle\DataflowType\Result;
 use CodeRhapsodie\DataflowBundle\DataflowType\Writer\WriterInterface;
+use CodeRhapsodie\DataflowBundle\Exceptions\InterruptedProcessingException;

 class Dataflow implements DataflowInterface
 {
--- a/src/DependencyInjection/CodeRhapsodieDataflowExtension.php
+++ b/src/DependencyInjection/CodeRhapsodieDataflowExtension.php
@@ -24,9 +24,5 @@ class CodeRhapsodieDataflowExtension extends Extension
            ->registerForAutoconfiguration(DataflowTypeInterface::class)
            ->addTag('coderhapsodie.dataflow.type')
        ;
-        $configuration = new Configuration();
-        $config = $this->processConfiguration($configuration, $configs);
-
-        $container->setParameter('coderhapsodie.dataflow.dbal_default_connection', $config['dbal_default_connection']);
    }
 }
--- a/src/DependencyInjection/Configuration.php
+++ b/src/DependencyInjection/Configuration.php
@@ -1,27 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace CodeRhapsodie\DataflowBundle\DependencyInjection;
-
-use Symfony\Component\Config\Definition\Builder\TreeBuilder;
-use Symfony\Component\Config\Definition\ConfigurationInterface;
-
-class Configuration implements ConfigurationInterface
-{
-    public function getConfigTreeBuilder()
-    {
-        $treeBuilder = new TreeBuilder();
-        $rootNode = $treeBuilder->root('code_rhapsodie_dataflow');
-
-        $rootNode
-            ->children()
-                ->scalarNode('dbal_default_connection')
-                    ->defaultValue('default')
-                ->end()
-            ->end()
-        ;
-
-        return $treeBuilder;
-    }
-}
--- a/src/Entity/Job.php
+++ b/src/Entity/Job.php
@@ -4,11 +4,14 @@ declare(strict_types=1);

 namespace CodeRhapsodie\DataflowBundle\Entity;

-use Symfony\Component\Validator\Constraints as Asserts;
+use Doctrine\ORM\Mapping as ORM;

 /**
 * Dataflow execution status.
 *
+ * @ORM\Entity(repositoryClass="CodeRhapsodie\DataflowBundle\Repository\JobRepository")
+ * @ORM\Table(name="cr_dataflow_job")
+ *
 * @codeCoverageIgnore
 */
 class Job
@@ -17,84 +20,83 @@ class Job
    const STATUS_RUNNING = 1;
    const STATUS_COMPLETED = 2;

-    private const KEYS = [
-        'id',
-        'status',
-        'label',
-        'dataflow_type',
-        'options',
-        'requested_date',
-        'scheduled_dataflow_id',
-        'count',
-        'exceptions',
-        'start_time',
-        'end_time',
-    ];
-
    /**
-     * @var int|null
+     * @var int
+     *
+     * @ORM\Id()
+     * @ORM\Column(type="integer")
+     * @ORM\GeneratedValue(strategy="AUTO")
     */
    private $id;

    /**
     * @var int
     *
-     * @Asserts\Range(min=0, max=2)
+     * @ORM\Column(type="integer")
     */
    private $status;

    /**
     * @var string|null
     *
-     * @Asserts\NotBlank()
-     * @Asserts\Length(min=1, max=255)
-     * @Asserts\Regex("#^[[:alnum:] ]+\z#u")
+     * @ORM\Column(type="string")
     */
    private $label;

    /**
     * @var string|null
     *
-     * @Asserts\NotBlank()
-     * @Asserts\Length(min=1, max=255)
-     * @Asserts\Regex("#^[[:alnum:]\\]+\z#u")
+     * @ORM\Column(type="string")
     */
    private $dataflowType;

    /**
     * @var array|null
+     *
+     * @ORM\Column(type="json")
     */
    private $options;

    /**
     * @var \DateTimeInterface|null
     *
-     * @Asserts\DateTime()
+     * @ORM\Column(type="datetime", nullable=true)
     */
    private $requestedDate;

    /**
-     * @var int|null
+     * @var ScheduledDataflow|null
+     *
+     * @ORM\ManyToOne(targetEntity="ScheduledDataflow", inversedBy="jobs")
+     * @ORM\JoinColumn(nullable=true)
     */
-    private $scheduledDataflowId;
+    private $scheduledDataflow;

    /**
     * @var int|null
+     *
+     * @ORM\Column(type="integer", nullable=true)
     */
    private $count;

    /**
     * @var array|null
+     *
+     * @ORM\Column(type="json", nullable=true)
     */
    private $exceptions;

    /**
     * @var \DateTimeInterface|null
+     *
+     * @ORM\Column(type="datetime", nullable=true)
     */
    private $startTime;

    /**
     * @var \DateTimeInterface|null
+     *
+     * @ORM\Column(type="datetime", nullable=true)
     */
    private $endTime;

@@ -111,72 +113,14 @@ class Job
            ->setOptions($scheduled->getOptions())
            ->setRequestedDate(clone $scheduled->getNext())
            ->setLabel($scheduled->getLabel())
-            ->setScheduledDataflowId($scheduled->getId());
-    }
-
-    public function __construct()
-    {
-        $this->count = 0;
-        $this->status = static::STATUS_PENDING;
-    }
-
-    public static function createFromArray(array $datas)
-    {
-        $lost = array_diff(static::KEYS, array_keys($datas));
-        if (count($lost) > 0) {
-            throw new \LogicException('The first argument of '.__METHOD__.'  must be contains: "'.implode(', ',
-                    $lost).'"');
-        }
-
-        $job = new self();
-        $job->id = null === $datas['id'] ? null : (int) $datas['id'];
-        $job->setStatus(null === $datas['status'] ? null : (int) $datas['status']);
-        $job->setLabel($datas['label']);
-        $job->setDataflowType($datas['dataflow_type']);
-        $job->setOptions($datas['options']);
-        $job->setRequestedDate($datas['requested_date']);
-        $job->setScheduledDataflowId(null === $datas['scheduled_dataflow_id'] ? null : (int) $datas['scheduled_dataflow_id']);
-        $job->setCount(null === $datas['count'] ? null : (int) $datas['count']);
-        $job->setExceptions($datas['exceptions']);
-        $job->setStartTime($datas['start_time']);
-        $job->setEndTime($datas['end_time']);
-
-        return $job;
-    }
-
-    public function toArray(): array
-    {
-        return [
-            'id' => $this->getId(),
-            'status' => $this->getStatus(),
-            'label' => $this->getLabel(),
-            'dataflow_type' => $this->getDataflowType(),
-            'options' => $this->getOptions(),
-            'requested_date' => $this->getRequestedDate(),
-            'scheduled_dataflow_id' => $this->getScheduledDataflowId(),
-            'count' => $this->getCount(),
-            'exceptions' => $this->getExceptions(),
-            'start_time' => $this->getStartTime(),
-            'end_time' => $this->getEndTime(),
-        ];
+            ->setScheduledDataflow($scheduled)
+        ;
    }

    /**
-     * @param int $id
-     *
-     * @return Job
+     * @return int
     */
-    public function setId(int $id): Job
-    {
-        $this->id = $id;
-
-        return $this;
-    }
-
-    /**
-     * @return int|null
-     */
-    public function getId(): ?int
+    public function getId(): int
    {
        return $this->id;
    }
@@ -282,21 +226,21 @@ class Job
    }

    /**
-     * @return int|null
+     * @return ScheduledDataflow|null
     */
-    public function getScheduledDataflowId(): ?int
+    public function getScheduledDataflow(): ?ScheduledDataflow
    {
-        return $this->scheduledDataflowId;
+        return $this->scheduledDataflow;
    }

    /**
-     * @param int|null $scheduledDataflowId
+     * @param ScheduledDataflow|null $scheduledDataflow
     *
     * @return Job
     */
-    public function setScheduledDataflowId(?int $scheduledDataflowId): Job
+    public function setScheduledDataflow(?ScheduledDataflow $scheduledDataflow): Job
    {
-        $this->scheduledDataflowId = $scheduledDataflowId;
+        $this->scheduledDataflow = $scheduledDataflow;

        return $this;
    }
--- a/src/Entity/ScheduledDataflow.php
+++ b/src/Entity/ScheduledDataflow.php
@@ -5,11 +5,14 @@ declare(strict_types=1);
 namespace CodeRhapsodie\DataflowBundle\Entity;

 use CodeRhapsodie\DataflowBundle\Validator\Constraints\Frequency;
-use Symfony\Component\Validator\Constraints as Asserts;
+use Doctrine\ORM\Mapping as ORM;

 /**
 * Schedule for a regular execution of a dataflow.
 *
+ * @ORM\Entity(repositoryClass="CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository")
+ * @ORM\Table(name="cr_dataflow_scheduled")
+ *
 * @codeCoverageIgnore
 */
 class ScheduledDataflow
@@ -21,104 +24,71 @@ class ScheduledDataflow
        '1 month',
    ];

-    private const KEYS = ['id', 'label', 'dataflow_type', 'options', 'frequency', 'next', 'enabled'];
-
    /**
-     * @var int|null
+     * @var int
+     *
+     * @ORM\Id()
+     * @ORM\Column(name="id", type="integer")
+     * @ORM\GeneratedValue(strategy="AUTO")
     */
    private $id;

    /**
     * @var string|null
     *
-     * @Asserts\NotBlank()
-     * @Asserts\Length(min=1, max=255)
-     * @Asserts\Regex("#^[[:alnum:] ]+\z#u")
+     * @ORM\Column(type="string")
     */
    private $label;

    /**
     * @var string|null
     *
-     * @Asserts\NotBlank()
-     * @Asserts\Length(min=1, max=255)
-     * @Asserts\Regex("#^[[:alnum:]\\]+\z#u")
+     * @ORM\Column(type="string")
     */
    private $dataflowType;

    /**
     * @var array|null
+     *
+     * @ORM\Column(type="json")
     */
    private $options;

    /**
     * @var string|null
     *
-     * @Asserts\NotBlank()
+     * @ORM\Column(type="string")
+     *
     * @Frequency()
     */
    private $frequency;

    /**
     * @var \DateTimeInterface|null
+     *
+     * @ORM\Column(type="datetime", nullable=true)
     */
    private $next;

    /**
     * @var bool|null
+     *
+     * @ORM\Column(type="boolean")
     */
    private $enabled;

-    public static function createFromArray(array $datas)
-    {
-        $lost = array_diff(static::KEYS, array_keys($datas));
-        if (count($lost) > 0) {
-            throw new \LogicException('The first argument of '.__METHOD__.'  must be contains: "'.implode(', ',
-                    $lost).'"');
-        }
-
-        $scheduledDataflow = new self();
-        $scheduledDataflow->id = null === $datas['id'] ? null : (int) $datas['id'];
-
-        $scheduledDataflow->setLabel($datas['label']);
-        $scheduledDataflow->setDataflowType($datas['dataflow_type']);
-        $scheduledDataflow->setOptions($datas['options']);
-        $scheduledDataflow->setFrequency($datas['frequency']);
-        $scheduledDataflow->setNext($datas['next']);
-        $scheduledDataflow->setEnabled(null === $datas['enabled'] ? null : (bool) $datas['enabled']);
-
-        return $scheduledDataflow;
-    }
-
-    public function toArray(): array
-    {
-        return [
-            'id' => $this->getId(),
-            'label' => $this->getLabel(),
-            'dataflow_type' => $this->getDataflowType(),
-            'options' => $this->getOptions(),
-            'frequency' => $this->getFrequency(),
-            'next' => $this->getNext(),
-            'enabled' => $this->getEnabled(),
-        ];
-    }
-
    /**
-     * @param int $id
+     * @var Job[]
     *
-     * @return ScheduledDataflow
+     * @ORM\OneToMany(targetEntity="Job", mappedBy="scheduledDataflow", cascade={"persist", "remove"})
+     * @ORM\OrderBy({"startTime" = "DESC"})
     */
-    public function setId(int $id): ScheduledDataflow
-    {
-        $this->id = $id;
-
-        return $this;
-    }
+    private $jobs;

    /**
-     * @return int|null
+     * @return int
     */
-    public function getId(): ?int
+    public function getId(): int
    {
        return $this->id;
    }
--- a/src/Factory/ConnectionFactory.php
+++ b/src/Factory/ConnectionFactory.php
@@ -1,35 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace CodeRhapsodie\DataflowBundle\Factory;
-
-use Symfony\Component\DependencyInjection\Container;
-
-/**
- * Class ConnectionFactory.
- *
- * @codeCoverageIgnore
- */
-class ConnectionFactory
-{
-    private $connectionName;
-
-    private $container;
-
-    public function __construct(Container $container, string $connectionName)
-    {
-        $this->connectionName = $connectionName;
-        $this->container = $container;
-    }
-
-    public function setConnectionName(string $connectionName)
-    {
-        $this->connectionName = $connectionName;
-    }
-
-    public function getConnection(): \Doctrine\DBAL\Driver\Connection
-    {
-        return $this->container->get(sprintf('doctrine.dbal.%s_connection', $this->connectionName));
-    }
-}
--- a/src/Manager/ScheduledDataflowManager.php
+++ b/src/Manager/ScheduledDataflowManager.php
@@ -8,25 +8,25 @@ use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
 use CodeRhapsodie\DataflowBundle\Entity\Job;
 use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
 use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
-use Doctrine\DBAL\Driver\Connection;
+use Doctrine\ORM\EntityManagerInterface;

 /**
 * Handles scheduled dataflows execution dates based on their frequency.
 */
 class ScheduledDataflowManager implements ScheduledDataflowManagerInterface
 {
+    /** @var EntityManagerInterface */
+    private $em;
+
    /** @var ScheduledDataflowRepository */
    private $scheduledDataflowRepository;

    /** @var JobRepository */
    private $jobRepository;

-    /** @var Connection */
-    private $connection;
-
-    public function __construct(Connection $connection, ScheduledDataflowRepository $scheduledDataflowRepository, JobRepository $jobRepository)
+    public function __construct(EntityManagerInterface $em, ScheduledDataflowRepository $scheduledDataflowRepository, JobRepository $jobRepository)
    {
-        $this->connection = $connection;
+        $this->em = $em;
        $this->scheduledDataflowRepository = $scheduledDataflowRepository;
        $this->jobRepository = $jobRepository;
    }
@@ -36,21 +36,16 @@ class ScheduledDataflowManager implements ScheduledDataflowManagerInterface
     */
    public function createJobsFromScheduledDataflows(): void
    {
-        $this->connection->beginTransaction();
-        try {
-            foreach ($this->scheduledDataflowRepository->findReadyToRun() as $scheduled) {
-                if (null !== $this->jobRepository->findPendingForScheduledDataflow($scheduled)) {
-                    continue;
-                }
-
-                $this->createPendingForScheduled($scheduled);
-                $this->updateScheduledDataflowNext($scheduled);
+        foreach ($this->scheduledDataflowRepository->findReadyToRun() as $scheduled) {
+            if (null !== $this->jobRepository->findPendingForScheduledDataflow($scheduled)) {
+                continue;
            }
-        } catch (\Throwable $e) {
-            $this->connection->rollBack();
-            throw $e;
+
+            $this->createPendingForScheduled($scheduled);
+            $this->updateScheduledDataflowNext($scheduled);
        }
-        $this->connection->commit();
+
+        $this->em->flush();
    }

    /**
@@ -67,7 +62,6 @@ class ScheduledDataflowManager implements ScheduledDataflowManagerInterface
        }

        $scheduled->setNext($next);
-        $this->scheduledDataflowRepository->save($scheduled);
    }

    /**
@@ -75,6 +69,6 @@ class ScheduledDataflowManager implements ScheduledDataflowManagerInterface
     */
    private function createPendingForScheduled(ScheduledDataflow $scheduled): void
    {
-        $this->jobRepository->save(Job::createFromScheduledDataflow($scheduled));
+        $this->em->persist(Job::createFromScheduledDataflow($scheduled));
    }
 }
--- a/src/Registry/DataflowTypeRegistry.php
+++ b/src/Registry/DataflowTypeRegistry.php
@@ -31,7 +31,7 @@ class DataflowTypeRegistry implements DataflowTypeRegistryInterface
            return $this->aliasesRegistry[$fqcnOrAlias];
        }

-        throw new UnknownDataflowTypeException($fqcnOrAlias);
+        throw new UnknownDataflowTypeException();
    }

    /**
--- a/src/Repository/InitFromDbTrait.php
+++ b/src/Repository/InitFromDbTrait.php
@@ -1,45 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace CodeRhapsodie\DataflowBundle\Repository;
-
-/**
- * @codeCoverageIgnore
- */
-trait InitFromDbTrait
-{
-    private function initDateTime(array $datas): array
-    {
-        foreach (static::FIELDS_TYPE as $key => $type) {
-            if ('datetime' === $type && null !== $datas[$key]) {
-                $datas[$key] = new \DateTime($datas[$key]);
-            }
-        }
-
-        return $datas;
-    }
-
-    private function initArray(array $datas): array
-    {
-        if (!is_array($datas['options'])) {
-            $datas['options'] = $this->strToArray($datas['options']);
-        }
-        if (array_key_exists('exceptions', $datas) && !is_array($datas['exceptions'])) {
-            $datas['exceptions'] = $this->strToArray($datas['exceptions']);
-        }
-
-        return $datas;
-    }
-
-    private function strToArray($value): array
-    {
-        if (null === $value) {
-            return [];
-        }
-
-        $array = json_decode($value, true);
-
-        return (false === $array) ? [] : $array;
-    }
-}
--- a/src/Repository/JobRepository.php
+++ b/src/Repository/JobRepository.php
@@ -4,176 +4,64 @@ declare(strict_types=1);

 namespace CodeRhapsodie\DataflowBundle\Repository;

-use CodeRhapsodie\DataflowBundle\Entity\Job;
 use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
-use Doctrine\DBAL\Driver\Connection;
-use Doctrine\DBAL\Query\QueryBuilder;
+use CodeRhapsodie\DataflowBundle\Entity\Job;
+use Doctrine\Common\Collections\Criteria;
+use Doctrine\ORM\EntityRepository;

 /**
 * Repository.
 *
 * @codeCoverageIgnore
 */
-class JobRepository
+class JobRepository extends EntityRepository
 {
-    use InitFromDbTrait;
-
-    public const TABLE_NAME = 'cr_dataflow_job';
-
-    private const FIELDS_TYPE = [
-        'id' => \PDO::PARAM_INT,
-        'status' => \PDO::PARAM_INT,
-        'label' => \PDO::PARAM_STR,
-        'dataflow_type' => \PDO::PARAM_STR,
-        'options' => \PDO::PARAM_STR,
-        'requested_date' => 'datetime',
-        'scheduled_dataflow_id' => \PDO::PARAM_INT,
-        'count' => \PDO::PARAM_INT,
-        'exceptions' => \PDO::PARAM_STR,
-        'start_time' => 'datetime',
-        'end_time' => 'datetime',
-    ];
-
-    /**
-     * @var \Doctrine\DBAL\Connection
-     */
-    private $connection;
-
-    /**
-     * JobRepository constructor.
-     *
-     * @param Connection $connection
-     */
-    public function __construct(Connection $connection)
-    {
-        $this->connection = $connection;
-    }
-
-    public function find(int $jobId)
-    {
-        $qb = $this->createQueryBuilder();
-        $qb
-            ->andWhere($qb->expr()->eq('id', $qb->createNamedParameter($jobId, \PDO::PARAM_INT)))
-        ;
-
-        return $this->returnFirstOrNull($qb);
-    }
-
    public function findOneshotDataflows(): iterable
    {
-        $qb = $this->createQueryBuilder();
-        $qb
-            ->andWhere($qb->expr()->isNull('scheduled_dataflow_id'))
-            ->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, \PDO::PARAM_INT)));
-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
-            return [];
-        }
-        while (false !== ($row = $stmt->fetch(\PDO::FETCH_ASSOC))) {
-            yield Job::createFromArray($this->initDateTime($this->initArray($row)));
-        }
+        return $this->findBy([
+            'scheduledDataflow' => null,
+            'status' => Job::STATUS_PENDING,
+        ]);
    }

    public function findPendingForScheduledDataflow(ScheduledDataflow $scheduled): ?Job
    {
-        $qb = $this->createQueryBuilder();
-        $qb
-            ->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($scheduled->getId(), \PDO::PARAM_INT)))
-            ->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, \PDO::PARAM_INT)));
-
-        return $this->returnFirstOrNull($qb);
+        return $this->findOneBy([
+            'scheduledDataflow' => $scheduled->getId(),
+            'status' => Job::STATUS_PENDING,
+        ]);
    }

    public function findNextPendingDataflow(): ?Job
    {
-        $qb = $this->createQueryBuilder();
-        $qb->andWhere($qb->expr()->lte('requested_date', $qb->createNamedParameter(new \DateTime(), 'datetime')))
-            ->andWhere($qb->expr()->eq('status', $qb->createNamedParameter(Job::STATUS_PENDING, \PDO::PARAM_INT)))
-            ->orderBy('requested_date', 'ASC')
+        $criteria = (new Criteria())
+            ->where(Criteria::expr()->lte('requestedDate', new \DateTime()))
+            ->andWhere(Criteria::expr()->eq('status', Job::STATUS_PENDING))
+            ->orderBy(['requestedDate' => Criteria::ASC])
            ->setMaxResults(1)
        ;

-        return $this->returnFirstOrNull($qb);
+        return $this->matching($criteria)->first() ?: null;
    }

    public function findLastForDataflowId(int $dataflowId): ?Job
    {
-        $qb = $this->createQueryBuilder();
-        $qb->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($dataflowId, \PDO::PARAM_INT)))
-            ->orderBy('requested_date', 'DESC')
-            ->setMaxResults(1)
-        ;
-
-        return $this->returnFirstOrNull($qb);
+        return $this->findOneBy(['scheduledDataflow' => $dataflowId], ['requestedDate' => 'desc']);
    }

    public function findLatests(): iterable
    {
-        $qb = $this->createQueryBuilder();
-        $qb
-            ->orderBy('requested_date', 'DESC')
-            ->setMaxResults(20);
-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
-            return [];
-        }
-        while (false !== ($row = $stmt->fetch(\PDO::FETCH_ASSOC))) {
-            yield Job::createFromArray($row);
-        }
+        return $this->findBy([], ['requestedDate' => 'desc'], 20);
    }

    public function findForScheduled(int $id): iterable
    {
-        $qb = $this->createQueryBuilder();
-        $qb->andWhere($qb->expr()->eq('scheduled_dataflow_id', $qb->createNamedParameter($id, \PDO::PARAM_INT)))
-            ->orderBy('requested_date', 'DESC')
-            ->setMaxResults(20);
-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
-            return [];
-        }
-        while (false !== ($row = $stmt->fetch(\PDO::FETCH_ASSOC))) {
-            yield Job::createFromArray($row);
-        }
+        return $this->findBy(['scheduledDataflow' => $id], ['requestedDate' => 'desc'], 20);
    }

    public function save(Job $job)
    {
-        $datas = $job->toArray();
-        unset($datas['id']);
-
-        if (is_array($datas['options'])) {
-            $datas['options'] = json_encode($datas['options']);
-        }
-        if (is_array($datas['exceptions'])) {
-            $datas['exceptions'] = json_encode($datas['exceptions']);
-        }
-
-        if (null === $job->getId()) {
-            $this->connection->insert(static::TABLE_NAME, $datas, static::FIELDS_TYPE);
-            $job->setId((int) $this->connection->lastInsertId());
-
-            return;
-        }
-        $this->connection->update(static::TABLE_NAME, $datas, ['id' => $job->getId()], static::FIELDS_TYPE);
-    }
-
-    public function createQueryBuilder($alias = null): QueryBuilder
-    {
-        $qb = $this->connection->createQueryBuilder();
-        $qb->select('*')
-            ->from(static::TABLE_NAME, $alias);
-
-        return $qb;
-    }
-
-    private function returnFirstOrNull(QueryBuilder $qb): ?Job
-    {
-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
-            return null;
-        }
-
-        return Job::createFromArray($this->initDateTime($this->initArray($stmt->fetch(\PDO::FETCH_ASSOC))));
+        $this->_em->persist($job);
+        $this->_em->flush();
    }
 }
--- a/src/Repository/ScheduledDataflowRepository.php
+++ b/src/Repository/ScheduledDataflowRepository.php
@@ -5,44 +5,16 @@ declare(strict_types=1);
 namespace CodeRhapsodie\DataflowBundle\Repository;

 use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
-use Doctrine\DBAL\Driver\Connection;
-use Doctrine\DBAL\Query\QueryBuilder;
+use Doctrine\Common\Collections\Criteria;
+use Doctrine\ORM\EntityRepository;

 /**
 * Repository for the ScheduledDataflow entity.
 *
 * @codeCoverageIgnore
 */
-class ScheduledDataflowRepository
+class ScheduledDataflowRepository extends EntityRepository
 {
-    use InitFromDbTrait;
-
-    public const TABLE_NAME = 'cr_dataflow_scheduled';
-
-    private const FIELDS_TYPE = [
-        'id' => \PDO::PARAM_INT,
-        'label' => \PDO::PARAM_STR,
-        'dataflow_type' => \PDO::PARAM_STR,
-        'options' => \PDO::PARAM_STR,
-        'frequency' => \PDO::PARAM_STR,
-        'next' => 'datetime',
-        'enabled' => \PDO::PARAM_BOOL,
-    ];
-    /**
-     * @var \Doctrine\DBAL\Connection
-     */
-    private $connection;
-
-    /**
-     * JobRepository constructor.
-     *
-     * @param Connection $connection
-     */
-    public function __construct(Connection $connection)
-    {
-        $this->connection = $connection;
-    }
-
    /**
     * Finds all enabled scheduled dataflows with a passed next run date.
     *
@@ -50,105 +22,42 @@ class ScheduledDataflowRepository
     */
    public function findReadyToRun(): iterable
    {
-        $qb = $this->createQueryBuilder();
-        $qb->andWhere($qb->expr()->lte('next', $qb->createNamedParameter(new \DateTime(), 'datetime')))
-            ->andWhere($qb->expr()->eq('enabled', 1))
-            ->orderBy('next', 'ASC')
-            ;
-
-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
-            return [];
-        }
-        while (false !== ($row = $stmt->fetch(\PDO::FETCH_ASSOC))) {
-            yield ScheduledDataflow::createFromArray($this->initDateTime($this->initArray($row)));
-        }
-    }
-
-    public function find(int $scheduleId): ?ScheduledDataflow
-    {
-        $qb = $this->createQueryBuilder();
-        $qb->andWhere($qb->expr()->eq('id', $qb->createNamedParameter($scheduleId, \PDO::PARAM_INT)))
-            ->setMaxResults(1)
+        $criteria = (new Criteria())
+            ->where(Criteria::expr()->lte('next', new \DateTime()))
+            ->andWhere(Criteria::expr()->eq('enabled', 1))
+            ->orderBy(['next' => Criteria::ASC])
        ;

-        return $this->returnFirstOrNull($qb);
+        return $this->matching($criteria);
    }

    public function findAllOrderedByLabel(): iterable
    {
-        $qb = $this->createQueryBuilder();
-        $qb->orderBy('label', 'ASC');
-
-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
-            return [];
-        }
-        while (false !== ($row = $stmt->fetch(\PDO::FETCH_ASSOC))) {
-            yield ScheduledDataflow::createFromArray($this->initDateTime($this->initOptions($row)));
-        }
+        return $this->findBy([], ['label' => 'asc']);
    }

    public function listAllOrderedByLabel(): array
    {
-        $query = $this->connection->createQueryBuilder()
-            ->from(static::TABLE_NAME, 'w')
-            ->select('w.id', 'w.label', 'w.enabled', 'w.next', 'max(j.start_time) as startTime')
-            ->leftJoin('w', JobRepository::TABLE_NAME, 'j', 'j.scheduled_dataflow_id = w.id')
+        $query = $this->createQueryBuilder('w')
+            ->select('w.id', 'w.label', 'w.enabled', 'w.next', 'max(j.startTime) as startTime')
+            ->leftJoin('w.jobs', 'j')
            ->orderBy('w.label', 'ASC')
            ->groupBy('w.id');

-        return $query->execute()->fetchAll(\PDO::FETCH_ASSOC);
+        return $query->getQuery()->execute();
    }

    public function save(ScheduledDataflow $scheduledDataflow)
    {
-        $datas = $scheduledDataflow->toArray();
-        unset($datas['id']);
-
-        if (is_array($datas['options'])) {
-            $datas['options'] = json_encode($datas['options']);
-        }
-
-        if (null === $scheduledDataflow->getId()) {
-            $this->connection->insert(static::TABLE_NAME, $datas, static::FIELDS_TYPE);
-            $scheduledDataflow->setId((int) $this->connection->lastInsertId());
-
-            return;
-        }
-        $this->connection->update(static::TABLE_NAME, $datas, ['id' => $scheduledDataflow->getId()], static::FIELDS_TYPE);
+        $this->_em->persist($scheduledDataflow);
+        $this->_em->flush();
    }

    public function delete(int $id): void
    {
-        $this->connection->beginTransaction();
-        try {
-            $this->connection->delete(JobRepository::TABLE_NAME, ['scheduled_dataflow_id' => $id]);
-            $this->connection->delete(static::TABLE_NAME, ['id' => $id]);
-        } catch (\Throwable $e) {
-            $this->connection->rollBack();
-            throw $e;
-        }
+        $dataflow = $this->find($id);

-        $this->connection->commit();
-    }
-
-    public function createQueryBuilder($alias = null): QueryBuilder
-    {
-        $qb = $this->connection->createQueryBuilder();
-        $qb->select('*')
-            ->from(static::TABLE_NAME, $alias);
-
-        return $qb;
-    }
-
-    private function returnFirstOrNull(QueryBuilder $qb): ?ScheduledDataflow
-    {
-        $stmt = $qb->execute();
-        if (0 === $stmt->rowCount()) {
-            return null;
-        }
-
-        return ScheduledDataflow::createFromArray($this->initDateTime($this->initArray($stmt->fetch(\PDO::FETCH_ASSOC))));
+        $this->_em->remove($dataflow);
+        $this->_em->flush();
    }
 }
--- a/src/Resources/config/services.yaml
+++ b/src/Resources/config/services.yaml
@@ -10,73 +10,53 @@ services:
            $registry: '@CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface'
            $scheduledDataflowRepository: '@CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository'
            $validator: '@validator'
-            $connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
        tags: ['console.command']

    CodeRhapsodie\DataflowBundle\Command\ChangeScheduleStatusCommand:
        arguments:
            $scheduledDataflowRepository: '@CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository'
-            $connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
        tags: ['console.command']

    CodeRhapsodie\DataflowBundle\Command\ExecuteDataflowCommand:
        arguments:
            $registry: '@CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface'
-            $connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
        tags: ['console.command']

    CodeRhapsodie\DataflowBundle\Command\JobShowCommand:
        arguments:
            $jobRepository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'
-            $connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
        tags: ['console.command']

    CodeRhapsodie\DataflowBundle\Command\RunPendingDataflowsCommand:
        arguments:
            $manager: '@CodeRhapsodie\DataflowBundle\Manager\ScheduledDataflowManagerInterface'
            $runner: '@CodeRhapsodie\DataflowBundle\Runner\PendingDataflowRunnerInterface'
-            $connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
        tags: ['console.command']

    CodeRhapsodie\DataflowBundle\Command\ScheduleListCommand:
        arguments:
            $scheduledDataflowRepository: '@CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository'
-            $connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
-        tags: ['console.command']
-
-    CodeRhapsodie\DataflowBundle\Command\SchemaCommand:
-        arguments:
-            $connectionFactory: '@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory'
        tags: ['console.command']

    CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository:
-        lazy: true
-        arguments: ['@coderhapsodie.dataflow.connection']
+        factory: ['@doctrine.orm.default_entity_manager', 'getRepository']
+        arguments: ['CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow']

    CodeRhapsodie\DataflowBundle\Repository\JobRepository:
-        lazy: true
-        arguments: ['@coderhapsodie.dataflow.connection']
-
-    coderhapsodie.dataflow.connection: "@coderhapsodie.dataflow.connection.internal"
-
-    coderhapsodie.dataflow.connection.internal:
-        lazy: true
-        class: Doctrine\DBAL\Connection
-        factory: ['@CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory', 'getConnection']
-
-    CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory:
-        arguments: ['@service_container', '%coderhapsodie.dataflow.dbal_default_connection%']
+        factory: ['@doctrine.orm.default_entity_manager', 'getRepository']
+        arguments: ['CodeRhapsodie\DataflowBundle\Entity\Job']

    CodeRhapsodie\DataflowBundle\Manager\ScheduledDataflowManagerInterface: '@CodeRhapsodie\DataflowBundle\Manager\ScheduledDataflowManager'
    CodeRhapsodie\DataflowBundle\Manager\ScheduledDataflowManager:
        arguments:
-            $connection: '@coderhapsodie.dataflow.connection'
+            $em: '@doctrine.orm.default_entity_manager'
            $scheduledDataflowRepository: '@CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository'
            $jobRepository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'

    CodeRhapsodie\DataflowBundle\Runner\PendingDataflowRunnerInterface: '@CodeRhapsodie\DataflowBundle\Runner\PendingDataflowRunner'
    CodeRhapsodie\DataflowBundle\Runner\PendingDataflowRunner:
        arguments:
+            $em: '@doctrine.orm.default_entity_manager'
            $repository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'
            $registry: '@CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface'
            $dispatcher: '@event_dispatcher'
--- a/src/Runner/PendingDataflowRunner.php
+++ b/src/Runner/PendingDataflowRunner.php
@@ -10,10 +10,14 @@ use CodeRhapsodie\DataflowBundle\Event\Events;
 use CodeRhapsodie\DataflowBundle\Event\ProcessingEvent;
 use CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface;
 use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
+use Doctrine\ORM\EntityManagerInterface;
 use Symfony\Component\EventDispatcher\EventDispatcherInterface;

 class PendingDataflowRunner implements PendingDataflowRunnerInterface
 {
+    /** @var EntityManagerInterface */
+    private $em;
+
    /** @var JobRepository */
    private $repository;

@@ -23,8 +27,9 @@ class PendingDataflowRunner implements PendingDataflowRunnerInterface
    /** @var EventDispatcherInterface */
    private $dispatcher;

-    public function __construct(JobRepository $repository, DataflowTypeRegistryInterface $registry, EventDispatcherInterface $dispatcher)
+    public function __construct(EntityManagerInterface $em, JobRepository $repository, DataflowTypeRegistryInterface $registry, EventDispatcherInterface $dispatcher)
    {
+        $this->em = $em;
        $this->repository = $repository;
        $this->registry = $registry;
        $this->dispatcher = $dispatcher;
@@ -56,7 +61,7 @@ class PendingDataflowRunner implements PendingDataflowRunnerInterface
            ->setStatus(Job::STATUS_RUNNING)
            ->setStartTime(new \DateTime())
        ;
-        $this->repository->save($job);
+        $this->em->flush();
    }

    /**
@@ -77,7 +82,7 @@ class PendingDataflowRunner implements PendingDataflowRunnerInterface
            ->setCount($result->getSuccessCount())
            ->setExceptions($exceptions)
        ;
-        $this->repository->save($job);
+        $this->em->flush();

        $this->dispatcher->dispatch(Events::AFTER_PROCESSING, new ProcessingEvent($job));
    }
--- a/src/SchemaProvider/DataflowSchemaProvider.php
+++ b/src/SchemaProvider/DataflowSchemaProvider.php
@@ -1,54 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace CodeRhapsodie\DataflowBundle\SchemaProvider;
-
-use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
-use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
-use Doctrine\DBAL\Schema\Schema;
-
-/**
- * Class JobSchemaProvider.
- *
- * @codeCoverageIgnore
- */
-class DataflowSchemaProvider
-{
-    public function createSchema()
-    {
-        $schema = new Schema();
-        $tableJob = $schema->createTable(JobRepository::TABLE_NAME);
-        $tableJob->addColumn('id', 'integer', array(
-            'autoincrement' => true,
-        ));
-        $tableJob->setPrimaryKey(array('id'));
-
-        $tableJob->addColumn('scheduled_dataflow_id', 'integer', ['notnull' => false]);
-        $tableJob->addColumn('status', 'integer', ['notnull' => true]);
-        $tableJob->addColumn('label', 'string', ['notnull' => true, 'length' => 255]);
-        $tableJob->addColumn('dataflow_type', 'string', ['notnull' => true, 'length' => 255]);
-        $tableJob->addColumn('options', 'json', ['notnull' => true]);
-        $tableJob->addColumn('requested_date', 'datetime', ['notnull' => false]);
-        $tableJob->addColumn('count', 'integer', ['notnull' => false]);
-        $tableJob->addColumn('exceptions', 'json', ['notnull' => false]);
-        $tableJob->addColumn('start_time', 'datetime', ['notnull' => false]);
-        $tableJob->addColumn('end_time', 'datetime', ['notnull' => false]);
-
-        $tableSchedule = $schema->createTable(ScheduledDataflowRepository::TABLE_NAME);
-        $tableSchedule->addColumn('id', 'integer', array(
-            'autoincrement' => true,
-        ));
-        $tableSchedule->setPrimaryKey(array('id'));
-        $tableSchedule->addColumn('label', 'string', ['notnull' => true, 'length' => 255]);
-        $tableSchedule->addColumn('dataflow_type', 'string', ['notnull' => true, 'length' => 255]);
-        $tableSchedule->addColumn('options', 'json', ['notnull' => true]);
-        $tableSchedule->addColumn('frequency', 'string', ['notnull' => true, 'length' => 255]);
-        $tableSchedule->addColumn('next', 'datetime', ['notnull' => false]);
-        $tableSchedule->addColumn('enabled', 'boolean', ['notnull' => true]);
-
-        $tableJob->addForeignKeyConstraint($tableSchedule, ['scheduled_dataflow_id'], ['id']);
-
-        return $schema;
-    }
-}