code-rhapsodie · jeremycr · Mar 31, 2026 · Feb 23, 2026 · Feb 23, 2026 · Mar 31, 2026
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,3 +1,7 @@
+# Version 5.5.0
+* Added cleanup commands
+* Updated for PHP 8.2
+
 # Version 5.4.1
 * Fix File Exceptions integration
 

diff --git a/README.md b/README.md
@@ -5,7 +5,7 @@ providing an easy way to create import / export dataflow.
 
 | Dataflow | Symfony                  | Support |
 |----------|--------------------------|---------|
-| 5.x      | 7.x                      | yes     |
+| 5.x      | ^7.3                     | yes     |
 | 4.x      | 3.4 \| 4.x \| 5.x \| 6.x | yes     |
 | 3.x      | 3.4 \| 4.x \| 5.x        | no      |
 | 2.x      | 3.4 \| 4.x               | no      |
@@ -243,7 +243,6 @@ implementing `DataflowTypeInterface`.
 
 Otherwise, manually add the tag `coderhapsodie.dataflow.type` in your dataflow type service configuration:
 
-```yaml
 ```yaml
 CodeRhapsodie\DataflowExemple\DataflowType\MyFirstDataflowType:
   tags:
@@ -598,6 +597,10 @@ the messenger component instead.
 
 `code-rhapsodie:dataflow:dump-schema` Generates schema create / update SQL queries
 
+`code-rhapsodie:dataflow:job:set-crashed` Jobs that have been in the "running" status for too long will be set in the "crashed" status.
+
+`code-rhapsodie:dataflow:job:cleanup` Remove old completed or crashed jobs
+
 ### Work with many databases
 
 All commands have a `--connection` option to define what Doctrine DBAL connection to use during execution.

diff --git a/Tests/DataflowType/AbstractDataflowTypeTest.php b/Tests/DataflowType/AbstractDataflowTypeTest.php
@@ -18,7 +18,7 @@ public function testProcess()
 
         $dataflowType = new class($label, $options, $values, $testCase) extends AbstractDataflowType
         {
-            public function __construct(private string $label, private array $options, private array $values, private TestCase $testCase)
+            public function __construct(private readonly string $label, private readonly array $options, private readonly array $values, private readonly TestCase $testCase)
             {
             }
 

diff --git a/composer.json b/composer.json
@@ -41,33 +41,35 @@
         }
     },
     "require": {
-        "php": "^8.0",
+        "php": "^8.2",
         "ext-json": "*",
         "doctrine/dbal": "^3.0||^4.0",
         "doctrine/doctrine-bundle": "^2.0",
         "monolog/monolog": "^2.0||^3.0",
         "psr/log": "^1.1||^2.0||^3.0",
-        "symfony/config": "^7.0",
-        "symfony/console": "^7.0",
-        "symfony/dependency-injection": "^7.0",
-        "symfony/event-dispatcher": "^7.0",
-        "symfony/http-kernel": "^7.0",
-        "symfony/lock": "^7.0",
-        "symfony/monolog-bridge": "^7.0",
-        "symfony/options-resolver": "^7.0",
-        "symfony/validator": "^7.0",
-        "symfony/yaml": "^7.0"
+        "symfony/config": "^7.3",
+        "symfony/console": "^7.3",
+        "symfony/dependency-injection": "^7.3",
+        "symfony/event-dispatcher": "^7.3",
+        "symfony/http-kernel": "^7.3",
+        "symfony/lock": "^7.3",
+        "symfony/monolog-bridge": "^7.3",
+        "symfony/options-resolver": "^7.3",
+        "symfony/validator": "^7.3",
+        "symfony/yaml": "^7.3"
     },
     "require-dev": {
         "amphp/amp": "^2.5",
+        "ekino/phpstan-banned-code": "^3.2",
         "friendsofphp/php-cs-fixer": "^3.75",
         "phpunit/phpunit": "^11",
         "portphp/portphp": "^1.9",
         "rector/rector": "^2.0",
-        "symfony/messenger": "^7.0"
+        "symfony/messenger": "^7.3"
     },
     "suggest": {
         "amphp/amp": "Provide asynchronous steps for your dataflows",
+        "league/flysystem":  "Allows Dataflow file exception mode, i.e. saving job exceptions outside the database",
         "portphp/portphp": "Provides generic readers, steps and writers for your dataflows.",
         "symfony/messenger": "Allows messenger mode, i.e. letting workers run jobs"
     },

diff --git a/phpstan.neon.dist b/phpstan.neon.dist
@@ -0,0 +1,6 @@
+includes:
+    - vendor/ekino/phpstan-banned-code/extension.neon
+parameters:
+    level: 5
+    paths:
+        - src/
diff --git a/rector.php b/rector.php
@@ -2,24 +2,20 @@
 
 declare(strict_types=1);
 
-use Rector\CodeQuality\Rector\Class_\InlineConstructorDefaultToPropertyRector;
 use Rector\Config\RectorConfig;
 use Rector\Set\ValueObject\LevelSetList;
 use Rector\Symfony\Set\SymfonySetList;
 
-return static function (RectorConfig $rectorConfig): void {
-    $rectorConfig->paths([
-        __DIR__ . '/src',
-        __DIR__ . '/Tests',
-    ]);
-
-    // register a single rule
-    $rectorConfig->rule(InlineConstructorDefaultToPropertyRector::class);
-
-    $rectorConfig->sets([
-        SymfonySetList::SYMFONY_70,
+return RectorConfig::configure()
+    ->withPaths([
+        __DIR__.'/src',
+        __DIR__.'/Tests',
+    ])
+    ->withComposerBased(doctrine: true, symfony: true)
+    ->withSets([
         SymfonySetList::SYMFONY_CODE_QUALITY,
         SymfonySetList::SYMFONY_CONSTRUCTOR_INJECTION,
-        LevelSetList::UP_TO_PHP_80,
-    ]);
-};
+        SymfonySetList::SYMFONY_73,
+        LevelSetList::UP_TO_PHP_82,
+    ])
+;
diff --git a/src/Command/AddScheduledDataflowCommand.php b/src/Command/AddScheduledDataflowCommand.php
@@ -9,46 +9,34 @@
 use CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface;
 use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
 use Symfony\Component\Console\Attribute\AsCommand;
-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\Attribute\Option;
 use Symfony\Component\Console\Style\SymfonyStyle;
 use Symfony\Component\Validator\Validator\ValidatorInterface;
 
 /**
  * @codeCoverageIgnore
  */
-#[AsCommand('code-rhapsodie:dataflow:schedule:add', 'Create a scheduled dataflow')]
-class AddScheduledDataflowCommand extends Command
+#[AsCommand('code-rhapsodie:dataflow:schedule:add', 'Create a scheduled dataflow', help: <<<'TXT'
+The <info>%command.name%</info> allows you to create a new scheduled dataflow.
+TXT)]
+final readonly class AddScheduledDataflowCommand
 {
     public function __construct(private DataflowTypeRegistryInterface $registry, private ScheduledDataflowRepository $scheduledDataflowRepository, private ValidatorInterface $validator, private ConnectionFactory $connectionFactory)
     {
-        parent::__construct();
     }
 
-    protected function configure(): void
-    {
-        $this
-            ->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('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');
-    }
-
-    protected function execute(InputInterface $input, OutputInterface $output): int
-    {
-        if ($input->getOption('connection') !== null) {
-            $this->connectionFactory->setConnectionName($input->getOption('connection'));
+    public function __invoke(
+        SymfonyStyle $io,
+        #[Option('Label of the scheduled dataflow')] ?string $label = null,
+        #[Option('Type of the scheduled dataflow (FQCN)')] ?string $type = null,
+        #[Option('Options of the scheduled dataflow (ex: {"option1": "value1", "option2": "value2"})')] ?string $options = null,
+        #[Option('Frequency of the scheduled dataflow')] ?string $frequency = null,
+        #[Option('Date for the first run of the scheduled dataflow (Y-m-d H:i:s)')] ?string $firstRun = null,
+        #[Option('State of the scheduled dataflow')] ?bool $enabled = null,
+        #[Option('Define the DBAL connection to use')] ?string $connection = null,
+    ): int {
+        if ($connection !== null) {
+            $this->connectionFactory->setConnectionName($connection);
         }
         $choices = [];
         $typeMapping = [];
@@ -57,36 +45,29 @@ protected function execute(InputInterface $input, OutputInterface $output): int
             $typeMapping[$dataflowType->getLabel()] = $fqcn;
         }
 
-        $io = new SymfonyStyle($input, $output);
-        $label = $input->getOption('label');
         if (!$label) {
             $label = $io->ask('What is the scheduled dataflow label?');
         }
-        $type = $input->getOption('type');
         if (!$type) {
             $selectedType = $io->choice('What is the scheduled dataflow type?', $choices);
             $type = $typeMapping[$selectedType];
         }
-        $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([])
             );
         }
-        $frequency = $input->getOption('frequency');
         if (!$frequency) {
             $frequency = $io->choice(
                 'What is the frequency for the scheduled dataflow?',
                 ScheduledDataflow::AVAILABLE_FREQUENCIES
             );
         }
-        $firstRun = $input->getOption('first_run');
         if (!$firstRun) {
             $firstRun = $io->ask('When is the first execution of the scheduled dataflow (format: Y-m-d H:i:s)?');
         }
-        $enabled = $input->getOption('enabled');
-        if (!$enabled) {
+        if ($enabled === null) {
             $enabled = $io->confirm('Enable the scheduled dataflow?');
         }
 

diff --git a/src/Command/ChangeScheduleStatusCommand.php b/src/Command/ChangeScheduleStatusCommand.php
@@ -7,63 +7,58 @@
 use CodeRhapsodie\DataflowBundle\Entity\ScheduledDataflow;
 use CodeRhapsodie\DataflowBundle\Factory\ConnectionFactory;
 use CodeRhapsodie\DataflowBundle\Repository\ScheduledDataflowRepository;
+use Symfony\Component\Console\Attribute\Argument;
 use Symfony\Component\Console\Attribute\AsCommand;
-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 Symfony\Component\Console\Attribute\Option;
 use Symfony\Component\Console\Style\SymfonyStyle;
 
 /**
  * @codeCoverageIgnore
  */
-#[AsCommand('code-rhapsodie:dataflow:schedule:change-status', 'Change schedule status')]
-class ChangeScheduleStatusCommand extends Command
+#[AsCommand('code-rhapsodie:dataflow:schedule:change-status', 'Change schedule status', help: <<<'TXT'
+The <info>%command.name%</info> command able you to change schedule status.
+TXT)]
+final readonly class ChangeScheduleStatusCommand
 {
     public function __construct(private ScheduledDataflowRepository $scheduledDataflowRepository, private ConnectionFactory $connectionFactory)
     {
-        parent::__construct();
     }
 
-    protected function configure(): void
-    {
-        $this
-            ->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');
-    }
-
-    protected function execute(InputInterface $input, OutputInterface $output): int
-    {
-        if ($input->getOption('connection') !== null) {
-            $this->connectionFactory->setConnectionName($input->getOption('connection'));
+    public function __invoke(
+        SymfonyStyle $io,
+        #[Argument('Id of the schedule')] int $scheduleId,
+        #[Option('Enable the schedule')] ?bool $enable = null,
+        #[Option('Disable the schedule')] ?bool $disable = null,
+        #[Option('Define the DBAL connection to use')] ?string $connection = null,
+    ): int {
+        if ($connection !== null) {
+            $this->connectionFactory->setConnectionName($connection);
         }
-        $io = new SymfonyStyle($input, $output);
+
         /** @var ScheduledDataflow|null $schedule */
-        $schedule = $this->scheduledDataflowRepository->find((int) $input->getArgument('schedule-id'));
+        $schedule = $this->scheduledDataflowRepository->find($scheduleId);
 
         if (!$schedule) {
-            $io->error(\sprintf('Cannot find scheduled dataflow with id "%d".', $input->getArgument('schedule-id')));
+            $io->error(\sprintf('Cannot find scheduled dataflow with id "%d".', $scheduleId));
 
             return 1;
         }
 
-        if ($input->getOption('enable') && $input->getOption('disable')) {
+        if ($enable !== null && $disable !== null) {
             $io->error('You cannot pass enable and disable options in the same time.');
 
             return 2;
         }
-        if (!$input->getOption('enable') && !$input->getOption('disable')) {
+        if ($enable === null && $disable === null) {
             $io->error('You must pass enable or disable option.');
 
             return 3;
         }
 
+        $enable = $enable ?? !$disable;
+
         try {
-            $schedule->setEnabled($input->getOption('enable'));
+            $schedule->setEnabled($enable);
             $this->scheduledDataflowRepository->save($schedule);
             $io->success(\sprintf('Schedule with id "%s" has been successfully updated.', $schedule->getId()));
         } catch (\Exception $e) {
-Original file line number
+Diff line change
@@ Expand Up / @@ -18,7 +18,7 @@ public function testProcess() @@
             $dataflowType = new class($label, $options, $values, $testCase) extends AbstractDataflowType
             {
-                public function __construct(private string $label, private array $options, private array $values, private TestCase $testCase)
+                public function __construct(private readonly string $label, private readonly array $options, private readonly array $values, private readonly TestCase $testCase)
                 {
                 }
@@ Expand Down @@