PHP code example of adamczykpiotr / laravel-dag-workflows

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

adamczykpiotr / laravel-dag-workflows example snippets

use AdamczykPiotr\DagWorkflows\Definitions\Task;
use AdamczykPiotr\DagWorkflows\Definitions\TaskGroup;
use AdamczykPiotr\DagWorkflows\Definitions\Workflow;

const TASK_FETCH_FEEDS = 'fetch_feeds';
const TASK_PARSE_CATALOGS = 'parse_catalogs';
const TASK_PARSE_ALBUMS = 'parse_albums';
const TASK_PARSE_IMAGES = 'parse_images';
const TASK_PARSE_METADATA = 'parse_metadata';

const TASK_SYNC_CATALOG_ALBUM_RELATIONS = 'sync_catalog_album_relations';
const TASK_SYNC_ALBUM_IMAGE_RELATIONS = 'sync_album_image_relations';
const TASK_SYNC_IMAGE_METADATA_RELATIONS = 'sync_image_metadata_relations';

$workflow = new Workflow(
    name: 'Image Import Pipeline',
    tasks: [
        new Task(
            name: TASK_FETCH_FEEDS,
            jobs: new DownloadFeedsJob(),
        ),

        new TaskGroup(
            tasks: [
                new Task(
                    name: TASK_PARSE_CATALOGS,
                    jobs: [
                        new ParseCatalogsJob('source-a'),
                        new ParseCatalogsJob('source-b'),
                    ],
                ),

                new Task(
                    name: TASK_PARSE_ALBUMS,
                    jobs: new ParseAlbumsJob(),
                ),

                new Task(
                    name: TASK_PARSE_IMAGES,
                    jobs: new ParseImagesJob(),
                ),

                new Task(
                    name: TASK_PARSE_METADATA,
                    jobs: new ParseImageMetadataJob(),
                ),
            ],
            dependsOn: TASK_FETCH_FEEDS,
        ),

        new Task(
            name: TASK_SYNC_CATALOG_ALBUM_RELATIONS,
            jobs: new SyncCatalogAlbumRelationsJob(),
            dependsOn: [TASK_PARSE_CATALOGS, TASK_PARSE_ALBUMS],
        ),

        new Task(
            name: TASK_SYNC_ALBUM_IMAGE_RELATIONS,
            jobs: new SyncAlbumImageRelationsJob(),
            dependsOn: [TASK_PARSE_ALBUMS, TASK_SYNC_CATALOG_ALBUM_RELATIONS],
        ),

        new Task(
            name: TASK_SYNC_IMAGE_METADATA_RELATIONS,
            jobs: new SyncImageMetadataRelationsJob(),
            dependsOn: [TASK_PARSE_IMAGES, TASK_SYNC_ALBUM_IMAGE_RELATIONS],
        ),
    ],
);

$model = $workflow->dispatch();
dump($model->id);

use AdamczykPiotr\DagWorkflows\Traits\HasWorkflowTracking;

class BuildDatasetJob implements ShouldQueue {
    use HasWorkflowTracking;

    public function rollbackStep(): void {
        // Undo whatever the failed attempt left behind.
        File::deleteDirectory($this->workDirectory());
    }

    public function handle(): void {
        // ... always starts from a clean slate
    }
}

use AdamczykPiotr\DagWorkflows\Models\Workflow;

$workflow = Workflow::find($id);

// Pause the entire workflow
$workflow->pause('Anomaly detected - awaiting manual review');

// Resume when ready
$workflow->resume();

// Or cancel if not recoverable
$workflow->cancel();

use AdamczykPiotr\DagWorkflows\Traits\HasWorkflowTracking;

class ScrapeDatasetJob implements ShouldQueue {
    use HasWorkflowTracking;

    public function handle(): void {
        $file = $this->download();

        if ($this->isUnchanged($file)) {
            $this->completeTaskEarly('source file unchanged'); // never returns
        }

        // ... continue as usual
    }
}

new Task(name: 'Import: Customers', jobs: new ImportCustomersJob()),
new ResolvableTask(
    name: 'Import: Orders',
    items: fn() => $regions,
    jobs: fn(string $region) => new ImportOrdersJob($region),
),
new Task(
    name: 'Import: Summary',
    jobs: new BuildImportSummaryJob(),
    dependsOn: 'Import: *', // waits for Import: Customers, Import: Orders and every Import: Orders:<region> task
),