Etl Core Bundle Laravel Package

What's new

enum support in configurationSchema

Fields declared in a step's configurationSchema (see v0.11.0) can now restrict the value to a fixed list of allowed values:

configurationSchema: [
    'mode' => ['type' => 'string', 'enum' => ['append', 'overwrite']],
],

• Comparison is strict (in_array($value, $enum, true)).
• Combinable with type — both constraints must pass.
• Also works standalone, without a type.
• A disallowed value throws InvalidStepConfigurationException with the list of allowed values in the message.

This is purely additive — no existing public signatures change. See the README for the full DSL.

Full changelog

v0.11.0...v0.12.0

What's new

configurationSchema on #[AsExecutableStep]

Steps can now declare the expected shape of their configuration, in addition to the existing human-readable configurationDescription:

#[AsExecutableStep(
    code: 'etl.loader.csv_file',
    configurationSchema: [
        'destination' => ['type' => 'string'],
        'withHeader'  => ['type' => 'boolean'],
    ],
)]

getConfigurationSchema() returns it verbatim, and it is now enforced at runtime: whenever setConfiguration() is called (e.g. by the pipeline runner), the given configuration is validated against the schema and an InvalidStepConfigurationException is thrown on mismatch.

Schema DSL, per field:

All 9 built-in steps (CSV/JSON/XML extractors and loaders, filter, data mapping, console command executor) now declare their configurationSchema.

This is purely additive — no existing public signatures change. See the README for the full DSL and an example.

Full changelog

v0.10.0...v0.11.0

What's new

Pipeline info and result in email notifications

EmailNotificationProvider now sends HTML instead of plain text. The email body includes:

• Pipeline details: id, name, scheduled/started/finished timestamps, duration
• The pipeline's final result, rendered inside a <pre> block (JSON pretty-printed for arrays/objects, HTML-escaped)

NotificationMessage gains a new optional field:

This is purely additive — no existing public signatures change. See the README for the full field list and a custom-provider example.

Full changelog

v0.9.0...v0.10.0

What's new

Pipeline purge scheduler

Old Pipeline runs can now be purged automatically so storage doesn't grow unbounded. A daily scheduled job (PurgeOldPipelines, registered in ETLScheduler) deletes pipelines in a terminal state (COMPLETED/FAILED) whose finishedAt is older than a configurable threshold, together with their PipelineHistory and StepHistory records. WorkflowStatistic/WorkflowExecutionStatistic records are intentionally left untouched so historical reporting survives purges.

Disabled by default.

# config/packages/boutdecode_etl_core.yaml
boutdecode_etl_core:
    purge:
        enabled: true                  # default: false
        retention_days: 30             # default: 30 — minimum: 1
        cron_expression: '0 3 * * *'   # default: '0 3 * * *' — validated at compile time

See the README for the full behavior and a Doctrine implementation example.

Breaking change

PipelinePersister, PipelineHistoryPersister and StepHistoryPersister each gain a new delete() method, and PipelineProvider gains findPurgeablePipelines(\DateTimeImmutable $before): array. Consuming applications implementing these interfaces must add the corresponding methods.

Full changelog

v0.8.1...v0.9.0

Fix

Follow-up to v0.7.0: notification preferences are now dedicated fields on Workflow instead of being read from the generic getConfiguration() array.

Breaking change

• Workflow gains three new methods: isNotifyOnSuccess(): bool, isNotifyOnFailure(): bool, getNotificationProviders(): ?array.
• AbstractWorkflow provides default implementations (false, false, null).
• WorkflowFactory::create() gains matching optional parameters: notifyOnSuccess, notifyOnFailure, notificationProviders.

$workflow = $workflowFactory->create(
    name: 'daily-import',
    notifyOnSuccess: false,
    notifyOnFailure: true,
    notificationProviders: ['email'], // optional — null defaults to every registered provider
);

Consuming applications implementing Workflow (or extending AbstractWorkflow) should add the corresponding columns/getters — see the README for a complete Doctrine entity example.

(Note: v0.8.0 was published and then deleted shortly after due to a tagging process issue — no functional difference between v0.8.0 and v0.8.1.)

Full changelog

v0.7.0...v0.8.1

What's new

Notifications domain

Get notified when a Workflow run succeeds and/or fails. After each pipeline run, NotificationMiddleware (priority -70, right after PipelineStatisticMiddleware) dispatches to every configured NotificationProvider. Sending a notification never fails the pipeline — errors are logged and swallowed.

Opting in from a Workflow

$workflow = $workflowFactory->create(
    name: 'daily-import',
    configuration: [
        'notifications' => [
            'on_success' => false,
            'on_failure' => true,
            'providers'  => ['email'], // optional — defaults to every registered provider
        ],
    ],
);

Built-in provider: Email

Ships with EmailNotificationProvider (via symfony/mailer). Recipients and sender are configured globally:

boutdecode_etl_core:
    notifications:
        email:
            from: '[email protected]'
            to: ['[email protected]']
            subject_prefix: '[ETL]'

Pluggable providers

Implement NotificationProvider and tag the class with #[AsNotificationProvider(code: '...')] — it is automatically registered (same pattern as ExecutableStep). See the README for a complete example.

Full changelog

v0.6.0...v0.7.0

Full Changelog: https://github.com/boutdecode/etl-core/compare/v0.5.0...v0.6.0

Full Changelog: https://github.com/boutdecode/etl-core/compare/v0.4.0...v0.5.0

What's new

Statistics domain

Automatic pre-aggregated execution metrics per pipeline, updated after every run.

New interfaces to implement (same pattern as Pipeline/Workflow)

Tracked metrics

• Total / success / failure counts
• Min, max, average execution duration (seconds)
• Last run date and status
• Success rate (getSuccessRate(): float)

New query handlers

• GetPipelineStatisticQuery($pipelineId) — returns the aggregated PipelineStatistic
• GetPipelineHistoriesQuery($pipelineId, $from, $to) — returns raw PipelineHistory records over a period (for temporal trend)

New entity to create

Extend AbstractPipelineStatistic and add Doctrine mappings — see the README for a complete example.

Full changelog

v0.3.1...v0.4.0

What's Changed

Added

• New scheduled state in the Pipeline state machine, inserted between pending and in_progress
• schedule() transition: pending → scheduled (triggered by ExecutePipelineHandler when a pipeline is picked up)
• reset transition now also accepts scheduled as a source state
• schedule() method on Pipeline interface and AbstractPipeline
• schedule() method on PipelineWorkflow interface and PipelineStateMachine

Changed

• IsPipelineExecutable now checks for SCHEDULED status instead of PENDING
• ExecutePipelineHandler transitions each pipeline to scheduled before dispatching ExecuteWorkflowCommand

Workflow

pending → (schedule) → scheduled → (start) → in_progress → (complete/fail) → completed/failed
                                                completed/failed → (reset) → pending
                                                         failed → (restart) → in_progress

What's new

• Add optional name property to Pipeline interface and AbstractPipeline
• Add ?string $name = null parameter to PipelineFactory::create() and PipelineFactory::createFromWorkflow()
• DefaultPlannedTaskRunner now names pipelines after their workflow: "<workflow> pipeline"

What's new

ETL — Executor step category

Add a new Executor step category alongside Extractor, Transformer and Loader.

New files

• ExecutorStep — interface extending ExecutableStep, exposes execute(string $command, array $arguments, Context $context, array $configuration): mixed
• AbstractExecutorStep — base class resolving command and arguments from the Context configuration, with fallback to the step's default configuration
• ConsoleCommandExecutorStep — built-in implementation running a Symfony console command via Application::run(), returns array{ exitCode: int, output: string }

Configuration keys for ConsoleCommandExecutorStep (etl.executor.console_command)

Dependencies

• symfony/console ^6.4\|\|^7.0 added as require-dev

🚀 Initial release

First stable release of ETLCoreBundle — a reusable Symfony Bundle providing a configurable ETL (Extract / Transform / Load) pipeline engine.

Features

Core

• Context DTO — the central pass-through object between steps
• Pipeline & Step abstractions (AbstractExtractorStep, AbstractTransformerStep, AbstractLoaderStep)
• PipelineStatus enum and pipeline lifecycle management

Built-in ETL steps

Execution engine

• DefaultPipelineRunner with middleware chains (pipeline-level & step-level)
• CycleLife middlewares for pipeline & step lifecycle tracking
• Symfony Workflow integration (pipeline_lifecycle state machine)
• Symfony Scheduler integration (PipelineScheduler with #[AsCronTask])

CQS layer

• CommandBus / QueryBus interfaces backed by Symfony Messenger
• AsyncCommand / AsyncQuery for opt-in async transport
• 3 Messenger buses: command.bus, query.bus, event.bus

Requirements

• PHP >= 8.2
• Symfony ^6.4 || ^7.0

Installation

composer require boutdecode/etl-core-bundle