Download the PHP package wizcodepl/laravel-pipe without Composer

laravel-pipe

Stage-based pipeline framework for batch ETL of record streams in Laravel — drivers, resolvers, per-stage stats.

If you have multiple product suppliers...

Each one ships their catalog in a different shape. Supplier A serves a JSON REST API with translations under name.{locale}. Supplier B emits an XML feed with parameters embedded as <table><tr><td> HTML inside the description. Supplier C hands you a CSV with questionable encoding. Supplier D needs OAuth + an MD5 hash signed with a timestamp to mint a token that expires every hour.

Every supplier has "color" — but A returns it structurally (features.color = "Black"), B has it inside the product name ("Acme Slim Tee L Black"), C buries it inside HTML (<tr><td>color</td><td>Onyx Mist</td></tr>), and D doesn't return color at all (you have to parse it from the model name).

The data is inconsistent. The same matte-black t-shirt will be:

• supplier A: "Black"
• supplier B: "BLACK"
• supplier C: "Jet" (single-word marketing color)
• supplier D: "Onyx Black" (manufacturer marketing color you want to collapse into your canonical "Black")

Some data is just missing. One supplier exposes material in the spec table. Another only ships a model name ("Merino Crew Neck" → you infer wool). A third returns nothing — you have to validate that the product fails the ProductType requirements and skip it.

The APIs are fragile — random 5xx responses, image-hosting timeouts, and new product categories you don't know about yet. Your sync either silently drops records or blows up mid-run leaving half-written state in the database.

This package is for that problem.

What you get

Core principles:

• The driver is lazy — loadItems() is a generator, --limit=10 stops fetching after 10 records.
• One Item per record — each record gets its own mutable bag, stages can't step on each other.
• Stage fails? The remaining stages for THAT record are skipped, the next record starts clean.
• PipelineRun tells you EXACTLY where the problem is: ValidateRequired Failed=300 → 300 records dropped because material was missing. StageFailureException::context is forwarded to onStageError so you log the exact SKU + the missing fields.

Quick start

Optionally publish the config file (only needed if you want to register drivers in config/laravel-pipe.php instead of using the package defaults):

Skeleton generator for a new supplier:

This creates:

Fill in the three TODOs, drop your stage classes into ApparelPipeline/Stages/, and register the driver in config/laravel-pipe.php:

Run:

Output:

Seven products landed in the database. Three failed at ValidateRequired — your sync_quality log channel will tell you per-SKU what was missing thanks to StageFailureException::context.

A stage is a tiny class

Stages typically come in three shapes inside an apparel pipeline:

Item — the mutable bag

Every record gets its own Item. Stages read and write through it:

The driver injects PipelineRun under the reserved key _run. Don't touch that from inside stages — the package owns it.

Failing a stage

Item::fail($message, $context = []) throws a StageFailureException. The abstract pipeline catches it, increments the stage's failed counter, calls onStageError($stage, $exception, $item), and stops the chain for that record.

Override onStageError in your concrete pipeline to wire it to your logger:

The ...$e->context spread merges per-stage diagnostics (manufacturer values, the unparseable raw string, the missing field list, etc.) into the log record.

Inconsistent data — Resolve* stages

The classic problem: two suppliers ship different strings for the same color.

Solution: a ResolveColor stage that maps every variant to the canonical Color::Black enum, with a substring fallback through a 50+ entry dictionary.

For attributes with a stable domain (size, material, pattern) use PHP enums with a label() method. For the long-tail of color names, substring matching with locale-specific fallbacks works well.

Validating required data

A ValidateRequired stage typically reads state.productType, checks which attributes that ProductType requires, and calls $item->fail(...) with the missing field list when anything is absent. The pipeline aborts for that record; the driver moves to the next one.

Per-stage invariant

Every record that entered a stage finishes in exactly one bucket. Failures on ResolveX mean unparseable input; failures on Persist* mean a DB constraint; failures on SyncImages mean the CDN was unreachable. The table after each sync tells you which one — no guessing.

How the engine works (in one paragraph)

AbstractPipeline::run() iterates stages() in order, wraps each in a try/catch (\Throwable), and on failure calls onStageError($stage, $e, $item) and stops the chain for that record. AbstractDriver::run(?int $limit) iterates loadItems() (a generator — --limit=10 actually stops the HTTP fetch after 10 records), routes each Item through the resolver, runs the chosen pipeline, and returns a PipelineRun with stats.

Commands

Generator: scaffolds three files in app/Pipes/{Name}/.

Run a sync. {driver} is the key from config/laravel-pipe.php. Output: per-stage stats table + Processed/Unrouted/Duration headline.

Failure records

Every stage failure is captured on PipelineRun (stage, message, context from Item::fail()). Read them with $run->getFailures() and wire them into whatever you already use — logs, alerts, audit tables.

To enrich each record (e.g. attach the SKU from item.baseData), override AbstractPipeline::buildFailureRecord() in your concrete pipeline:

End-to-end test

Mock your API client or hit the live one — the package doesn't get in the way.

Skip from CI when network/credentials are unavailable: vendor/bin/phpunit --exclude-group=e2e.

Requirements

• PHP 8.2+
• Laravel 11 / 12 (only required for RunCommand, MakePipeCommand and PipeServiceProvider — the artisan layer)

The core (Item, AbstractPipeline, AbstractDriver, PipelineRun, StageFailureException) is plain PHP 8.2. Laravel is only needed for the optional artisan + service-provider layer.

About Wizcode

Wizcode is an e-commerce agency specialised in Lunar. We design and ship B2B, B2C, and marketplace platforms on the Laravel + Lunar stack — from custom checkouts and supplier syncs to multi-channel pricing, PIM workflows, and headless storefronts.

Our open-source contributions to the Lunar ecosystem:

• wizcodepl/lunar-product-schemas — migration-style schema builder for Lunar product types and attributes.
• wizcodepl/laravel-pipe — stage-based pipeline framework for batch ETL of supplier feeds (used in production for catalog ingestion).

Contact us: https://wizcode.pl

License

MIT