upline/pipeline-database-synchronizer

There is no license information available for the latest version (1.3.0) of this package.

1.3.0 2025-09-12 13:05 UTC

Unknown License

• vail

README

This is a set of tools to set up a synchronization pipeline. It allows to sync data by chunks to prevent memory consumption.

Installation

composer require upline/pipeline-database-synchronizer

Usage

<?php
use Upline\PipelineDatabaseSynchronizer\Base\DataWrap;
use Upline\PipelineDatabaseSynchronizer\Base\Ref;
use Upline\PipelineDatabaseSynchronizer\Base\StaticRef;
use Upline\PipelineDatabaseSynchronizer\SynchronizerFacade;
use Upline\PipelineDatabaseSynchronizer\Base\DatabaseDriver;


class YourDatabase implements DatabaseDriver 
{
    public function find(string $table, array $columns): array;

    public function findMany(string $table, array $idBags): array;

    public function update(string $table, array $idBag, array $data): array;

    public function insert(string $table, array $idBag, array $data): array;
} 

$exampleData = [
  [
     'db_id' => '1',
     'name' => 'Chair',
     'material' => 'Oak',
     'store' => [
        'id' => 1
     ]
  ]
];

$appFacade = new SynchronizerFacade(new YourDatabase());

// Search in materials by name
$exampleMapper = $appFacade->makeDatabaseMapper('materials', [
    'name' => Ref::make(DataWrap::fn(), 'material')
]);

// Fill products table
$productSyncer = $appFacade->makeBatchSyncer(
    table: 'products',
    ids: [
        'id' => ''
    ],
    columns: [
        // Static data
        'quantity' => StaticRef::make(100000),
        // Ref to the field from the original data
        'name' => 'name',
        // Ref to the mapper field
        'material_id' => Ref::make($exampleMapper, 'id'),
        // Ref to the nested field
        'store_id' => Ref::make(DataWrap::fn(), 'store', 'id'),
    ]
)
    // cast field after retrieving from db
    ->addCast('id', fn($v) => (int)$v)


// Build the pipeline.
$pipeline = $appFacade->makePipelineBuilder()
    ->map($exampleMapper)
    ->syncBatch($productSyncer, batchSize: 500)
    ->getResult();

foreach ($exampleData as $item) {
    $pipeline->processRaw($item);
}
$pipeline->end();




//// Create an ArraySplitter for locales.
//$localeSplitter = $appFacade->makeArraySplitter(Ref::make(null, 'locales'));
//
//$attributeSplitter = $appFacade->makeListSplitter(fields: [
//    1 => Ref::make(null, 'attr1'),
//    2 => Ref::make(null, 'attr2'),
//    17 => Ref::make(null, 'attr3'),
//]);