Skip to content

Commit 20600f6

Browse files
oliverde8oliverde8
committed
Added support for PHP-ETL 1.1 & Symfony 6.
1 parent 2634e5f commit 20600f6

File tree

11 files changed

+142
-61
lines changed

11 files changed

+142
-61
lines changed

‎CHANGELOG.md‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,12 @@
1+
# 1.1.0
2+
3+
- :star2: Support for php-etl 1.1 new operations has been added.
4+
- :star2: Support for symfony 6.0 has been added.
5+
- :star2: Support for using input parameters in operation options has been added.
6+
- :collision: All operations are no longer built during DI compilation. This should improve peformance of cache warmup.
7+
- :collision: Support for Symfony 4.4 has been dropped.
8+
- :collision: Support for php 7.4 has been dropped.
9+
110
# 1.0.4
211
- :wrench: Fix logs not being saved in some conditions.
312
-

‎Command/GetDefinitionCommand.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ protected function configure()
3636
protected function execute(InputInterface $input, OutputInterface $output)
3737
{
3838
$chainName = $input->getArgument("name");
39-
$definition = $this->chainProcessorsManager->getDefinition($chainName);
39+
$definition = $this->chainProcessorsManager->getRawDefinition($chainName);
4040

4141
echo Yaml::dump($definition, 4);
4242
return 0;

‎DependencyInjection/Compiler/ChainCompiler.php‎

Lines changed: 2 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -23,16 +23,7 @@ public function process(ContainerBuilder $container)
2323

2424
$chainProcessorManager = $container->getDefinition(ChainProcessorsManager::class);
2525

26-
foreach ($chainsArray as $chainName => $chain) {
27-
$chainDefinition = $container->register("oliverde8.etl.chain.$chainName", ChainProcessor::class);
28-
$chainDefinition->setFactory(new Reference(ChainFactory::class));
29-
$chainDefinition->setArgument('$config', $chain);
30-
31-
// Using public for performance. We should use lazy loading later on which would give same performance.
32-
$chainDefinition->setPublic(true);
33-
$chainDefinition->setShared(false);
34-
}
35-
36-
$chainProcessorManager->setArgument('$definitions', $chainsString);
26+
$chainProcessorManager->setArgument('$definitions', $chainsArray);
27+
$chainProcessorManager->setArgument('$rawDefinitions', $chainsString);
3728
}
3829
}

‎DependencyInjection/Compiler/ChainParameterCompiler.php‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,14 +30,14 @@ public function process(ContainerBuilder $container)
3030
}
3131
}
3232

33-
foreach ($etlFiles as $filename => $etlFile) {
33+
foreach ($etlFiles as $etlFile) {
3434
// Register file so that when it's modified in dev mode symfony empty caches automatially.
3535
$container->fileExists($etlFile);
3636

3737
$etlName = str_replace(".yml", "", $etlFile->getBasename());
3838
$ymlContent = file_get_contents($etlFile);
3939

40-
$definitionsArray[$etlName] = Yaml::parse($ymlContent)['chain'];
40+
$definitionsArray[$etlName] = Yaml::parse($ymlContent);
4141
$definitionsString[$etlName] = $ymlContent;
4242
}
4343

‎EventSubscriber/EtlExecutionEventSubscriber.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@ public function setChainDetails(BeforeEntityPersistedEvent $event)
5050
return;
5151
}
5252

53-
$definition = $this->chainProcessorManager->getDefinition($entity->getName());
53+
$definition = $this->chainProcessorManager->getRawDefinition($entity->getName());
5454
$entity->setDefinition($definition);
5555
$entity->setStatus(EtlExecution::STATUS_WAITING);
5656
}

‎Factory/ChainFactory.php‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55

66

77
use Oliverde8\Component\PhpEtl\ChainBuilder;
8+
use Oliverde8\Component\PhpEtl\ChainProcessor;
89

910
class ChainFactory
1011
{
@@ -20,8 +21,8 @@ public function __construct(ChainBuilder $chainBuilder)
2021
$this->chainBuilder = $chainBuilder;
2122
}
2223

23-
public function __invoke($config)
24+
public function create($config, array $inputOptions, int $maxAsynchronousItems): ChainProcessor
2425
{
25-
return $this->chainBuilder->buildChainProcessor($config);
26+
return $this->chainBuilder->buildChainProcessor($config, $inputOptions, $maxAsynchronousItems);
2627
}
2728
}

‎Model/LoggerProxy.php‎

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -23,48 +23,48 @@ public function setExecutionContext(ExecutionContext $executionContext): void
2323
$this->executionContext = $executionContext;
2424
}
2525

26-
public function emergency($message, array $context = array())
26+
public function emergency(string|\Stringable $message, array $context = array()): void
2727
{
2828
$this->logger->emergency($message, $this->executionContext->getLoggerContext($context));
2929
}
3030

31-
public function alert($message, array $context = array())
31+
public function alert(string|\Stringable $message, array $context = array()): void
3232
{
3333
$this->logger->alert($message, $this->executionContext->getLoggerContext($context));
3434
}
3535

36-
public function critical($message, array $context = array())
36+
public function critical(string|\Stringable $message, array $context = array()): void
3737
{
3838
$this->logger->critical($message, $this->executionContext->getLoggerContext($context));
3939
}
4040

41-
public function error($message, array $context = array())
41+
public function error(string|\Stringable $message, array $context = array()): void
4242
{
4343
$this->logger->error($message, $this->executionContext->getLoggerContext($context));
4444
}
4545

46-
public function warning($message, array $context = array())
46+
public function warning(string|\Stringable $message, array $context = array()): void
4747
{
4848
$this->logger->warning($message, $this->executionContext->getLoggerContext($context));
4949
}
5050

51-
public function notice($message, array $context = array())
51+
public function notice(string|\Stringable $message, array $context = array()): void
5252
{
5353
$this->logger->notice($message, $this->executionContext->getLoggerContext($context));
5454
}
5555

56-
public function info($message, array $context = array())
56+
public function info(string|\Stringable $message, array $context = array()): void
5757
{
5858
$this->logger->info($message, $this->executionContext->getLoggerContext($context));
5959
}
6060

61-
public function debug($message, array $context = array())
61+
public function debug(string|\Stringable $message, array $context = array()): void
6262
{
6363
$this->logger->debug($message, $this->executionContext->getLoggerContext($context));
6464
}
6565

66-
public function log($level, $message, array $context = array())
66+
public function log($level, string|\Stringable $message, array $context = array()): void
6767
{
6868
$this->logger->log($level, $message, $this->executionContext->getLoggerContext($context));
6969
}
70-
}
70+
}

‎Resources/config/service-operation-factories.yml‎

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,50 @@
11
services:
2+
#
3+
# Low Level Operation Builders
4+
#
5+
Oliverde8\Component\PhpEtl\Builder\Factories\ChainSplitFactory:
6+
class: Oliverde8\Component\PhpEtl\Builder\Factories\ChainSplitFactory
7+
autowire: true
8+
arguments:
9+
$operation: 'split'
10+
$class: 'Oliverde8\Component\PhpEtl\ChainOperation\ChainSplitOperation'
11+
tags:
12+
- { name: etl.operation-factory }
13+
14+
#
15+
# Extraction Operation Builders
16+
#
17+
Oliverde8\Component\PhpEtl\Builder\Factories\Extract\JsonExtractFactory:
18+
class: Oliverde8\Component\PhpEtl\Builder\Factories\Extract\JsonExtractFactory
19+
autowire: true
20+
arguments:
21+
$operation: 'json-read'
22+
$class: 'Oliverde8\Component\PhpEtl\ChainOperation\Extract\JsonExtractOperation'
23+
tags:
24+
- { name: etl.operation-factory }
25+
26+
Oliverde8\Component\PhpEtl\Builder\Factories\Extract\CsvExtractFactory:
27+
class: Oliverde8\Component\PhpEtl\Builder\Factories\Extract\CsvExtractFactory
28+
autowire: true
29+
arguments:
30+
$operation: 'csv-read'
31+
$class: 'Oliverde8\Component\PhpEtl\ChainOperation\Extract\CsvExtractOperation'
32+
tags:
33+
- { name: etl.operation-factory }
34+
35+
36+
#
37+
# Transformation Operation Builders.
38+
#
39+
Oliverde8\Component\PhpEtl\Builder\Factories\Transformer\FilterDataFactory:
40+
class: Oliverde8\Component\PhpEtl\Builder\Factories\Transformer\FilterDataFactory
41+
autowire: true
42+
arguments:
43+
$operation: 'filter'
44+
$class: 'Oliverde8\Component\PhpEtl\ChainOperation\Transformer\FilterDataOperation'
45+
tags:
46+
- { name: etl.operation-factory }
47+
248
Oliverde8\Component\PhpEtl\Builder\Factories\Transformer\RuleTransformFactory:
349
class: Oliverde8\Component\PhpEtl\Builder\Factories\Transformer\RuleTransformFactory
450
autowire: true
@@ -8,6 +54,24 @@ services:
854
tags:
955
- { name: etl.operation-factory }
1056

57+
Oliverde8\Component\PhpEtl\Builder\Factories\Transformer\SplitItemFactory:
58+
class: Oliverde8\Component\PhpEtl\Builder\Factories\Transformer\SplitItemFactory
59+
autowire: true
60+
arguments:
61+
$operation: 'item-split'
62+
$class: 'Oliverde8\Component\PhpEtl\ChainOperation\Transformer\SplitItemOperation'
63+
tags:
64+
- { name: etl.operation-factory }
65+
66+
Oliverde8\Component\PhpEtl\Builder\Factories\Transformer\SimpleHttpOperationFactory:
67+
class: Oliverde8\Component\PhpEtl\Builder\Factories\Transformer\SimpleHttpOperationFactory
68+
autowire: true
69+
arguments:
70+
$operation: 'http'
71+
$class: 'Oliverde8\Component\PhpEtl\ChainOperation\Transformer\SimpleHttpOperation'
72+
tags:
73+
- { name: etl.operation-factory }
74+
1175
Oliverde8\Component\PhpEtl\Builder\Factories\Grouping\SimpleGroupingFactory:
1276
class: Oliverde8\Component\PhpEtl\Builder\Factories\Grouping\SimpleGroupingFactory
1377
autowire: true
@@ -17,6 +81,10 @@ services:
1781
tags:
1882
- { name: etl.operation-factory }
1983

84+
#
85+
# Load Operation Builders.
86+
#
87+
2088
Oliverde8\Component\PhpEtl\Builder\Factories\Loader\CsvFileWriterFactory:
2189
class: Oliverde8\Component\PhpEtl\Builder\Factories\Loader\CsvFileWriterFactory
2290
autowire: true
@@ -25,3 +93,12 @@ services:
2593
$class: 'Oliverde8\Component\PhpEtl\ChainOperation\Loader\FileWriterOperation'
2694
tags:
2795
- { name: etl.operation-factory }
96+
97+
Oliverde8\Component\PhpEtl\Builder\Factories\Loader\JsonFileWriterFactory:
98+
class: Oliverde8\Component\PhpEtl\Builder\Factories\Loader\JsonFileWriterFactory
99+
autowire: true
100+
arguments:
101+
$operation: 'json-write'
102+
$class: 'Oliverde8\Component\PhpEtl\ChainOperation\Loader\FileWriterOperation'
103+
tags:
104+
- { name: etl.operation-factory }

‎Security/EtlExecutionVoter.php‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ class EtlExecutionVoter extends Voter
1313
const DASHBOARD = 'dashboard';
1414
const DOWNLOAD = 'download';
1515

16-
protected function supports($attribute, $subject)
16+
protected function supports(string $attribute, mixed $subject): bool
1717
{
1818
if ($subject == EtlExecution::class) {
1919
return true;
@@ -26,7 +26,7 @@ protected function supports($attribute, $subject)
2626
return false;
2727
}
2828

29-
protected function voteOnAttribute($attribute, $subject, TokenInterface $token)
29+
protected function voteOnAttribute(string $attribute, mixed $subject, TokenInterface $token): bool
3030
{
3131
return true;
3232
}

‎Services/ChainProcessorsManager.php‎

Lines changed: 29 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -3,45 +3,41 @@
33
namespace Oliverde8\PhpEtlBundle\Services;
44

55
use Oliverde8\Component\PhpEtl\ChainProcessor;
6-
use Oliverde8\Component\PhpEtl\Exception\ChainOperationException;
7-
use Oliverde8\Component\PhpEtl\Item\DataItem;
8-
use Oliverde8\Component\PhpEtl\Item\DataItemInterface;
96
use Oliverde8\PhpEtlBundle\Entity\EtlExecution;
107
use Oliverde8\PhpEtlBundle\Exception\UnknownChainException;
8+
use Oliverde8\PhpEtlBundle\Factory\ChainFactory;
119
use Oliverde8\PhpEtlBundle\Repository\EtlExecutionRepository;
12-
use Psr\Container\ContainerInterface;
13-
use Psr\Log\LoggerInterface;
1410

1511
class ChainProcessorsManager
1612
{
17-
protected ContainerInterface $container;
18-
1913
protected EtlExecutionRepository $etlExecutionRepository;
20-
2114
protected LoggerFactory $loggerFactory;
22-
15+
protected ChainFactory $chainFactory;
2316
protected array $definitions;
17+
protected array $rewDefinitions;
2418

2519
public function __construct(
26-
ContainerInterface $container,
2720
EtlExecutionRepository $etlExecutionRepository,
2821
LoggerFactory $loggerFactory,
29-
array $definitions
22+
ChainFactory $chainFactory,
23+
array $definitions,
24+
array $rawDefinitions
3025
) {
31-
$this->container = $container;
3226
$this->etlExecutionRepository = $etlExecutionRepository;
3327
$this->loggerFactory = $loggerFactory;
28+
$this->chainFactory = $chainFactory;
3429
$this->definitions = $definitions;
30+
$this->rewDefinitions = $rawDefinitions;
3531
}
3632

3733
/**
3834
* @throws UnknownChainException
3935
*/
40-
public function getDefinition(string $chainName): string
36+
public function getRawDefinition(string $chainName): string
4137
{
42-
if (!isset($this->definitions[$chainName])) {
38+
if (!isset($this->rewDefinitions[$chainName])) {
4339
$alternatives = [];
44-
foreach (array_keys($this->definitions) as $knownId) {
40+
foreach (array_keys($this->rewDefinitions) as $knownId) {
4541
$lev = levenshtein($chainName, $knownId);
4642
if ($lev <= \strlen($chainName) / 3 || str_contains($knownId, $chainName)) {
4743
$alternatives[] = $knownId;
@@ -51,18 +47,25 @@ public function getDefinition(string $chainName): string
5147
throw new UnknownChainException("Unknown chain '$chainName', did you mean: " . implode(", ", $alternatives));
5248
}
5349

54-
return $this->definitions[$chainName];
50+
return $this->rewDefinitions[$chainName];
5551
}
5652

57-
public function getDefinitions(): array
53+
public function getRewDefinitions(): array
5854
{
59-
return $this->definitions;
55+
return $this->rewDefinitions;
6056
}
6157

62-
public function getProcessor(string $chainName): ChainProcessor
58+
public function getProcessor(string $chainName, array $options): ChainProcessor
6359
{
64-
// TODO Think about either creating the processor & runtime or injecting them into the constructor like the definitions.
65-
return $this->container->get("oliverde8.etl.chain.$chainName");
60+
$this->getRawDefinition($chainName);
61+
$definition = $this->definitions[$chainName];
62+
$chain = $definition['chain'];
63+
$maxAsynchronousItems = $definition['maxAsynchronousItems'] ?? 20;
64+
$defaultOptions = $definition['defaultOptions'] ?? [];
65+
66+
$options = array_merge($defaultOptions, $options);
67+
68+
return $this->chainFactory->create($chain, $options, $maxAsynchronousItems);
6669
}
6770

6871
/**
@@ -76,7 +79,7 @@ public function getProcessor(string $chainName): ChainProcessor
7679
*/
7780
public function execute(string $chainName, iterable $iterator, array $params)
7881
{
79-
$definition = $this->getDefinition($chainName);
82+
$definition = $this->getRawDefinition($chainName);
8083

8184
$inputData = ["Iterator! Can't show input data"];
8285
if (is_array($iterator)) {
@@ -95,7 +98,7 @@ public function execute(string $chainName, iterable $iterator, array $params)
9598
* Execute a chain from it's entity.
9699
*
97100
*/
98-
public function executeFromEtlEntity(EtlExecution $execution, iterable $iterator = null)
101+
public function executeFromEtlEntity(EtlExecution $execution, iterable $iterator = null): void
99102
{
100103
$chainName = $execution->getName();
101104
$logger = $this->loggerFactory->get($execution);
@@ -108,8 +111,8 @@ public function executeFromEtlEntity(EtlExecution $execution, iterable $iterator
108111
$this->etlExecutionRepository->save($execution);
109112

110113
// Build the processor.
111-
$processor = $this->getProcessor($chainName);
112114
$params = json_decode($execution->getInputOptions(), true);
115+
$processor = $this->getProcessor($chainName, $params);
113116

114117
if (is_null($iterator)) {
115118
$iterator = new \ArrayIterator(json_decode($execution->getInputData(), true));
@@ -128,7 +131,7 @@ public function executeFromEtlEntity(EtlExecution $execution, iterable $iterator
128131
$execution = $this->etlExecutionRepository->find($execution->getId());
129132
$execution->setFailTime(new \DateTime());
130133
$execution->setStatus(EtlExecution::STATUS_FAILURE);
131-
$execution->setErrorMessage($this->getFullExeptionTrace($exception));
134+
$execution->setErrorMessage($this->getFullExceptionTrace($exception));
132135
throw $exception;
133136
} finally {
134137
$execution->setEndTime(new \DateTime());
@@ -138,7 +141,7 @@ public function executeFromEtlEntity(EtlExecution $execution, iterable $iterator
138141
}
139142
}
140143

141-
protected function getFullExeptionTrace(\Throwable $exception)
144+
protected function getFullExceptionTrace(\Throwable $exception): string
142145
{
143146
$message = '';
144147
do {

0 commit comments

Comments
 (0)