PHPackages                             ivanfuhr/ingestor - PHPackages - PHPackages  [Skip to content](#main-content)[PHPackages](/)[Directory](/)[Categories](/categories)[Trending](/trending)[Leaderboard](/leaderboard)[Changelog](/changelog)[Analyze](/analyze)[Collections](/collections)[Log in](/login)[Sign up](/register)

1. [Directory](/)
2. /
3. [PDF & Document Generation](/categories/documents)
4. /
5. ivanfuhr/ingestor

ActiveLibrary[PDF & Document Generation](/categories/documents)

ivanfuhr/ingestor
=================

v1.0.14(2w ago)0241MITPHPPHP ^8.2CI passing

Since Jun 15Pushed 2w agoCompare

[ Source](https://github.com/ivanfuhr/ingestor)[ Packagist](https://packagist.org/packages/ivanfuhr/ingestor)[ RSS](/packages/ivanfuhr-ingestor/feed)WikiDiscussions main Synced 1w ago

READMEChangelog (10)Dependencies (8)Versions (18)Used By (0)

[ ![Ingestor](art/header.png)](https://github.com/ivanfuhr/ingestor)Ingestor
========

[](#ingestor)

 [![Total Downloads](https://camo.githubusercontent.com/4cdb47896e221336ec1933662c89c5c54d8ba831fc84f1e91aa7523266a7f980/68747470733a2f2f696d672e736869656c64732e696f2f7061636b61676973742f64742f6976616e667568722f696e676573746f72)](https://packagist.org/packages/ivanfuhr/ingestor) [![Latest Stable Version](https://camo.githubusercontent.com/3e82164c4546681aea543ad15fd43c8a9ca56befe24006f5c4d19c40f2d3d222/68747470733a2f2f696d672e736869656c64732e696f2f7061636b61676973742f762f6976616e667568722f696e676573746f72)](https://packagist.org/packages/ivanfuhr/ingestor) [![License](https://camo.githubusercontent.com/447db6a9f23ea368c7ff6fbafeb8ea761da1a9c0979bafa8976b33554adb624c/68747470733a2f2f696d672e736869656c64732e696f2f7061636b61676973742f6c2f6976616e667568722f696e676573746f72)](https://packagist.org/packages/ivanfuhr/ingestor)

Ingestor is a PHP library for **safe, auditable data imports** with isolated staging, atomic release, and an extensible pipeline.

Data enters through a source driver, is transformed into mutations by a definition, is applied in an isolated stage by a persistence driver, and is only then promoted to production — safely and atomically.

> **Requires [PHP 8.2+](https://php.net/releases/)**, the **PDO** extension (persistence), and the **zip** and **xml** extensions (XLSX source).

Installation
------------

[](#installation)

⚡️ Get started by requiring the package using [Composer](https://getcomposer.org):

```
composer require ivanfuhr/ingestor
```

Quick Start
-----------

[](#quick-start)

```
use Ivanfuhr\Ingestor\Ingestor;
use Ivanfuhr\Ingestor\Driver\Persistence\PostgresDriver;
use Ivanfuhr\Ingestor\Driver\Source\CsvDriver;

$ingestor = Ingestor::make(
    persistence: new PostgresDriver($pdo),
    source: new CsvDriver(),
);

$import = $ingestor
    ->for(CustomerImport::class)
    ->from('/path/to/customers.csv')
    ->import();

if ($import->hasFailures()) {
    foreach ($import->failures() as $failure) {
        // inspect validation or persistence failures
    }

    $import->rollback();

    return;
}

$import->release();
```

Table of Contents
-----------------

[](#table-of-contents)

- [Architecture](#-architecture)
- [Definitions & Schema](#-definitions--schema)
- [Context](#-context)
- [Validation](#-validation)
- [Persistence Failures](#-persistence-failures)
- [Hooks](#-hooks)
- [Metrics](#-metrics)
- [Testing Utilities](#-testing-utilities)
- [PostgreSQL Driver](#-postgresql-driver)
- [CSV Driver](#-csv-driver)
- [XLSX Driver](#-xlsx-driver)
- [Development](#-development)
- [Community](#community)
- [License](#license)

### 🏗️ Architecture

[](#️-architecture)

Ingestor separates four responsibilities:

```
Source
    ↓
Source Driver
    ↓
Iterable
    ↓
Definition (prepare → validate → map)
    ↓
Dataset (mutations)
    ↓
Persistence Driver
    ↓
Stage (isolated)
    ↓
Release (atomic promotion)

```

DriverResponsibilityImplementations**Source**Turns a source into input rows`CsvDriver`, `XlsxDriver`**Persistence**Creates staging, persists mutations, and releases`PostgresDriver`Drivers are injected at construction time. The import pipeline never needs to know how data is read or written.

```
$ingestor = Ingestor::make(
    persistence: new PostgresDriver($pdo),
    source: new CsvDriver(),
);
```

**Why:** Keeps reading, transformation, and persistence independent — each piece can be swapped or tested in isolation.

---

### 📋 Definitions & Schema

[](#-definitions--schema)

A **Definition** describes an import. It declares structure via `Schema` and transforms each row into write intentions via `Dataset`.

```
use Ivanfuhr\Ingestor\Contract\Definition;
use Ivanfuhr\Ingestor\Contract\Context;
use Ivanfuhr\Ingestor\Dataset\Dataset;
use Ivanfuhr\Ingestor\Row\Row;
use Ivanfuhr\Ingestor\Schema\Schema;
use Ivanfuhr\Ingestor\Stage\EmptyStage;
use Ivanfuhr\Ingestor\Stage\PrefilledStage;
use Ivanfuhr\Ingestor\Conflict\UpdateOnConflict;

final class CustomerImport implements Definition
{
    public function schema(): Schema
    {
        return Schema::make()
            ->dataset('customers')
                ->using(PrefilledStage::class)
                ->onConflict(UpdateOnConflict::by('document'))
            ->dataset('addresses')
                ->using(EmptyStage::class);
    }

    public function map(Row $row, Context $context): Dataset
    {
        return Dataset::make()
            ->insert('customers', [
                'document' => $row->string('cpf'),
                'name' => $row->string('name'),
            ])
            ->insert('addresses', [
                'document' => $row->string('cpf'),
                'city' => $row->string('city'),
            ]);
    }
}
```

#### Stage Strategies

[](#stage-strategies)

StrategyBehavior`EmptyStage`Dataset starts empty`PrefilledStage`Dataset starts with a copy of existing data (ideal for incremental updates)`PrefilledStage` copies production rows into the staging table, including explicit surrogate keys. With `PostgresDriver`, serial/identity sequences on the staging table are synchronized to the highest copied value so new inserts that omit the key do not collide with prefilled rows.

Use `PrefilledStage::withoutSequenceSync()` when every insert provides an explicit surrogate key and you do not want the driver to adjust staging sequences:

```
return Schema::make()
    ->dataset('customers')
        ->using(PrefilledStage::class) // default: sync sequences after prefill
        ->onConflict(UpdateOnConflict::by('document'))
    ->dataset('legacy_accounts')
        ->using(PrefilledStage::withoutSequenceSync()) // explicit IDs only
        ->commit();
```

`DatasetBuilder::using()` accepts a stage strategy class name or a configured instance.

#### Conflict Strategies

[](#conflict-strategies)

Declared in the Schema and translated by the persistence driver:

```
UpdateOnConflict::by('document');
UpdateOnConflict::by('document', DuplicateInBatch::FirstWins);
IgnoreOnConflict::by('document');
ReplaceOnConflict::by('document');
FailOnConflict::by('document');
```

`UpdateOnConflict` and `ReplaceOnConflict` deduplicate rows that share the same conflict key within a single insert batch before executing `ON CONFLICT DO UPDATE`. By default, the **last row wins** (`DuplicateInBatch::LastWins`). This prevents PostgreSQL error `ON CONFLICT DO UPDATE command cannot affect row a second time`, which occurs when duplicate keys appear in the same multi-row `INSERT` — common with `PrefilledStage` incremental imports, but not caused by the stage strategy itself.

`DuplicateInBatch`Behavior`LastWins` (default)Keep the last occurrence of each conflict key in the batch`FirstWins`Keep the first occurrence`Fail`Abort the batch and report failures for duplicate keysA **Stage** is an isolated ingestion environment. Nothing touches production until `release()` is called.

```
Import
└── Stage
    ├── customers (staging table)
    └── addresses (staging table)

```

**Why:** One row can produce zero, one, or many mutations across multiple datasets — without coupling business logic to SQL.

---

### 🗂️ Context

[](#️-context)

Shared storage available throughout an import. Use it to preload ID maps, caches, and reference data so `map()` stays pure and fast.

```
use Ivanfuhr\Ingestor\Contract\Preparable;

final class OrderImport implements Definition, Preparable
{
    public function prepare(Context $context): void
    {
        $context->put('customers', Customer::pluck('id', 'document')->all());
    }

    public function map(Row $row, Context $context): Dataset
    {
        return Dataset::make()->insert('orders', [
            'customer_id' => $context->get('customers', $row->string('document')),
            'total' => $row->float('total'),
        ]);
    }
}
```

**Why:** Avoids N+1 queries during import. I/O belongs in `prepare()`; `map()` should be a pure `Row + Context → Dataset` transformation.

---

### ✅ Validation

[](#-validation)

Row validation is optional and runs before mapping. Implement `ValidatesRows` on your definition:

```
use Ivanfuhr\Ingestor\Contract\ValidatesRows;
use Ivanfuhr\Ingestor\Row\Row;
use Ivanfuhr\Ingestor\Validation\Failure;

final class CustomerImport implements Definition, ValidatesRows
{
    public function validate(Row $row, Context $context): iterable
    {
        if ($row->missing('document')) {
            yield Failure::error('document')
                ->message('Document is required.');
        }

        if ($row->missing('phone')) {
            yield Failure::warning('phone')
                ->message('Phone number is empty.');
        }
    }
}
```

SeverityDefault behaviorOverride`ERROR`Row is skipped — not mapped or persisted`->continueRow()` keeps the row in the pipeline`WARNING`Recorded, but the row continues through the pipeline`->skipRow()` drops the row```
yield Failure::warning('phone')
    ->skipRow()
    ->message('Phone is required for this import.');

yield Failure::error('legacy_code')
    ->continueRow()
    ->message('Legacy code is invalid but row can still be imported.');
```

Failures are available after import:

```
$import->failures();
$import->hasFailures();
```

**Why:** Invalid rows are caught early, before any database writes, with full reporting for audits and reprocessing.

---

### 🚨 Persistence Failures

[](#-persistence-failures)

Database errors (NOT NULL, FOREIGN KEY, UNIQUE, etc.) are exposed through the same `Failure` mechanism, with additional context:

- `line()` — original source line number
- `dataset()` — affected dataset
- `data()` — row data
- `cause()` — underlying exception

Failures **do not** trigger an automatic rollback. You decide between `release()` and `rollback()`.

```
$import = $ingestor
    ->for(CustomerImport::class)
    ->from($file)
    ->import();

if ($import->hasFailures()) {
    foreach ($import->failures() as $failure) {
        dump([
            'line' => $failure->line(),
            'dataset' => $failure->dataset(),
            'message' => $failure->message(),
            'data' => $failure->data(),
        ]);
    }

    $import->rollback();
    return;
}

$import->release();
```

#### SQL Failure Modes

[](#sql-failure-modes)

`PostgresDriver` supports configurable failure diagnosis:

```
use Ivanfuhr\Ingestor\Driver\Persistence\SqlFailureMode;

new PostgresDriver($pdo, chunkSize: 500, failureMode: SqlFailureMode::Diagnostic);
```

ModePriority`Fast`Throughput — records batch failure when a bulk INSERT fails`Diagnostic`Traceability — subdivides the batch to isolate the failing row**Why:** Every mutation inherits its source row context, so persistence errors remain traceable even at scale.

---

### 🔗 Hooks

[](#-hooks)

High-level lifecycle hooks for auditing, metrics, notifications, and external integrations. They run a fixed number of times regardless of row volume.

```
beforeImport()
    ↓
prepare()
    ↓
validate() → map() → persist()
    ↓
afterImport()
    ↓
release()
    ↓
beforeRelease() → promote stage → afterRelease()

```

InterfaceWhenTypical use`BeforeImport`Before import startsTimers, logging, audit trail`AfterImport`After all rows processed, before releaseMetrics, reports, notifications`BeforeRelease`Immediately before promotionFinal checks, manual approval`AfterRelease`After promotionCache invalidation, external sync`BeforeRelease` can block publication:

```
use Ivanfuhr\Ingestor\Exception\CannotRelease;

public function beforeRelease(ImportedImport $import): void
{
    if ($import->hasFailures()) {
        throw CannotRelease::because('Import contains unresolved failures.');
    }
}
```

**Why:** Integrate with the outside world without per-row callbacks that would destroy throughput.

---

### 📊 Metrics

[](#-metrics)

Read-only metrics collected during import. Available whether you release or rollback. Every dataset declared in the schema is included in the per-dataset breakdown, even when it produced no mutations.

```
$metrics = $import->metrics();

$metrics->startedAt();
$metrics->finishedAt();
$metrics->duration();

$metrics->rows();          // rows processed
$metrics->importedRows();  // rows imported successfully
$metrics->failedRows();    // rows with failures
$metrics->mutations();     // mutations produced

foreach ($metrics->datasets() as $dataset) {
    $dataset->name();
    $dataset->stageStrategy();     // e.g. PrefilledStage::class
    $dataset->onConflict();        // ConflictType or null
    $dataset->onConflictColumns(); // e.g. ['document']
    $dataset->mutations();
    $dataset->persisted();
    $dataset->failures();
}
```

Failures answer *what* and *why*. Metrics answer *how much*, *how long*, and *how each dataset was configured*.

**Why:** Every import becomes observable — performance, throughput, per-dataset breakdowns, and schema configuration (staging strategy and conflict handling) without affecting the pipeline.

---

### 🧪 Testing Utilities

[](#-testing-utilities)

Test definitions in isolation — no database, no CSV or XLSX files, no external infrastructure.

#### Asserting the Schema

[](#asserting-the-schema)

```
use Ivanfuhr\Ingestor\Ingestor;

Ingestor::test(CustomerImport::class)
    ->assertDataset('customers')
    ->assertStage(PrefilledStage::class)
    ->assertUpdateOnConflict('document');
```

#### Asserting `map()`

[](#asserting-map)

```
Ingestor::test(CustomerImport::class)
    ->withContext(['customers' => ['12345678901' => 1]])
    ->map(['cpf' => '12345678901', 'name' => 'Ada', 'city' => 'SP'])
    ->assertInserted('customers', [
        'document' => '12345678901',
        'name' => 'Ada',
    ])
    ->assertDatasetCount('addresses', 1);
```

#### Asserting Validation

[](#asserting-validation)

```
Ingestor::test(CustomerImport::class)
    ->map(['document' => null])
    ->assertFailure(field: 'document', message: 'Document is required.')
    ->assertFailureCount(1);
```

#### Asserting the Full Pipeline

[](#asserting-the-full-pipeline)

```
Ingestor::test(CustomerImport::class)
    ->fromRows([
        ['cpf' => '1', 'name' => 'Ada', 'city' => 'SP'],
        ['cpf' => '2', 'name' => 'Bob', 'city' => 'RJ'],
    ])
    ->import()
    ->assertRows(2)
    ->assertImportedRows(2)
    ->assertFailedRows(0)
    ->assertMutations(4);
```

**Why:** Definitions should be fully testable with fast, deterministic tests — safe to refactor without spinning up infrastructure.

---

### 🐘 PostgreSQL Driver

[](#-postgresql-driver)

`PostgresDriver` creates staging tables, inserts data in configurable batches, and atomically promotes staging to production.

```
use Ivanfuhr\Ingestor\Driver\Persistence\PostgresDriver;
use Ivanfuhr\Ingestor\Driver\Persistence\SqlFailureMode;

$driver = new PostgresDriver(
    pdo: $pdo,
    chunkSize: 500,
    failureMode: SqlFailureMode::Fast,
);
```

The driver introspects production tables to build matching staging tables and applies conflict strategies from the Schema via `ON CONFLICT`.

**Why:** Staging + atomic swap gives you a safe rollback window before data ever reaches production.

---

### 📄 CSV Driver

[](#-csv-driver)

`CsvDriver` reads CSV files with a header row and yields `RowContext` objects with line numbers and associative data.

```
use Ivanfuhr\Ingestor\Driver\Source\CsvDriver;

$ingestor = Ingestor::make($persistence, new CsvDriver());
```

Skip rows where every column is blank — common in CSV exports with trailing empty lines or `,,,` separators:

```
new CsvDriver(ignoreEmptyRows: true);
```

A row is considered empty when all values are `null`, `''`, or whitespace only. Disabled by default.

**Why:** Line numbers flow through the entire pipeline, enabling precise failure reporting back to the source file.

---

### 📊 XLSX Driver

[](#-xlsx-driver)

`XlsxDriver` reads Excel `.xlsx` files with a header row and yields `RowContext` objects with line numbers and associative data — same API as `CsvDriver`.

It has **zero Composer dependencies**: the file is opened as a ZIP archive and parsed incrementally with `XMLReader`, keeping memory use low for large sheets.

```
use Ivanfuhr\Ingestor\Driver\Source\XlsxDriver;
use Ivanfuhr\Ingestor\Driver\Source\XlsxSheet;

// First worksheet (default)
$ingestor = Ingestor::make($persistence, new XlsxDriver());

$ingestor
    ->for(CustomerImport::class)
    ->from('/path/to/customers.xlsx')
    ->import();

// Select a worksheet by name or zero-based index
new XlsxDriver(XlsxSheet::byName('Orders'));
new XlsxDriver(XlsxSheet::byIndex(1));

// Skip rows where every column is blank
new XlsxDriver(ignoreEmptyRows: true);
new XlsxDriver(XlsxSheet::byName('Orders'), ignoreEmptyRows: true);
```

FeatureSupportShared stringsYesInline stringsYesBooleans and formula cached valuesYesExcel serial datesReturned as raw numbersMultiple sheetsOne sheet per driver instanceIgnore empty rowsOptional via `ignoreEmptyRows: true`**Why:** Spreadsheet imports get the same DX and traceability as CSV — header-based associative rows and Excel row numbers for failure reporting — without pulling in PhpSpreadsheet or similar.

---

### 🛠️ Development

[](#️-development)

```
composer test          # PHPUnit
composer lint          # PHP-CS-Fixer (check)
composer lint:fix      # PHP-CS-Fixer (fix)
composer phpstan       # Static analysis
composer rector        # Automated refactoring
```

Community
---------

[](#community)

- [Contributing](CONTRIBUTING.md)
- [Security Policy](SECURITY.md)
- [Code of Conduct](CODE_OF_CONDUCT.md)
- [Changelog](CHANGELOG.md)

License
-------

[](#license)

**Ingestor** was created by **[Ivan Führ](https://github.com/ivanfuhr)** under the **[MIT license](LICENSE)**.

###  Health Score

47

—

FairBetter than 93% of packages

Maintenance98

Actively maintained with recent releases

Popularity16

Limited adoption so far

Community8

Small or concentrated contributor base

Maturity55

Maturing project, gaining track record

 Bus Factor1

Top contributor holds 90.5% of commits — single point of failure

How is this calculated?**Maintenance (25%)** — Last commit recency, latest release date, and issue-to-star ratio. Uses a 2-year decay window.

**Popularity (30%)** — Total and monthly downloads, GitHub stars, and forks. Logarithmic scaling prevents top-heavy scores.

**Community (15%)** — Contributors, dependents, forks, watchers, and maintainers. Measures real ecosystem engagement.

**Maturity (30%)** — Project age, version count, PHP version support, and release stability.

###  Release Activity

Cadence

Every ~2 days

Total

15

Last Release

14d ago

### Community

Maintainers

![](https://www.gravatar.com/avatar/27f53ee9cc9ae53da93c2f9546885f5fa6baf88694a2d20d5bf81f18c593101f?d=identicon)[ivanfuhr](/maintainers/ivanfuhr)

---

Top Contributors

[![ivanfuhr](https://avatars.githubusercontent.com/u/72282810?v=4)](https://github.com/ivanfuhr "ivanfuhr (38 commits)")[![cursoragent](https://avatars.githubusercontent.com/u/199161495?v=4)](https://github.com/cursoragent "cursoragent (4 commits)")

---

Tags

csvetlimportphp

###  Code Quality

TestsPHPUnit

Static AnalysisPHPStan, Rector

Code StylePHP CS Fixer

Type Coverage Yes

### Embed Badge

![Health badge](/badges/ivanfuhr-ingestor/health.svg)

```
[![Health](https://phpackages.com/badges/ivanfuhr-ingestor/health.svg)](https://phpackages.com/packages/ivanfuhr-ingestor)
```

###  Alternatives

[tarfin-labs/easy-pdf

Makes pdf processing easy.

1719.9k](/packages/tarfin-labs-easy-pdf)[akeneo-labs/excel-connector-bundle

Akeneo PIM Excel connector bundle

166.4k](/packages/akeneo-labs-excel-connector-bundle)

PHPackages © 2026

[Directory](/)[Categories](/categories)[Trending](/trending)[Changelog](/changelog)[Analyze](/analyze)