data_pipelines-1.x-dev/data_pipelines.post_update.php

<?php

/**
 * @file
 * Post update functions for data_pipelines module.
 */

use Drupal\Core\Site\Settings;
use Drupal\data_pipelines\Entity\DatasetInterface;

/**
 * Add machine names to all DataSet entities.
 */
function data_pipelines_post_update_add_machine_name(&$sandbox = NULL) {
  $storage = \Drupal::entityTypeManager()->getStorage('data_pipelines');
  $iteration_size = (int) Settings::get('entity_update_batch_size', 50);
  if (empty($sandbox['progress'])) {
    $sandbox['progress'] = 0;
    $sandbox['max'] = (int) $storage->getQuery()
      ->accessCheck(FALSE)
      ->count()
      ->execute();
  }

  $start = $sandbox['progress'];
  $end = min($sandbox['max'], $start + $iteration_size);

  $ids = $storage->getQuery()
    ->sort('id')
    ->accessCheck(FALSE)
    ->range($start, $end)
    ->execute();
  $datasets = $storage->loadMultiple(array_values($ids));
  $value_taken = function ($id, $machine_name) use ($storage) {
    return (bool) $storage->getQuery()
      ->condition('id', $id, '<>')
      ->condition('machine_name', $machine_name)
      ->range(0, 1)
      ->count()
      ->accessCheck(FALSE)
      ->execute();
  };
  foreach ($datasets as $id => $dataset) {
    $machine_name = preg_replace('@[^a-z0-9_]+@', '_', trim(mb_strtolower($dataset->label())));
    while ($value_taken($id, $machine_name)) {
      $machine_name .= '_' . mt_rand(0, 9);
    }
    $dataset->machine_name->value = $machine_name;
    $dataset->save();
  }
  if ($sandbox['max'] > 0 && $end < $sandbox['max']) {
    $sandbox['progress'] = $end;
    $sandbox['#finished'] = ($end - 1) / $sandbox['max'];
  }
  else {
    $sandbox['#finished'] = 1;
  }
}

/**
 * Recreate all the existing indexes.
 */
function data_pipelines_post_update_delete_elastic_indexes(&$sandbox = NULL) {
  // Deprecated.
}

/**
 * Create destination entities.
 */
function data_pipelines_post_update_destinations(&$sandbox = NULL) {
  $data_set_destination = NULL;
  $entity_type_manager = \Drupal::entityTypeManager();

  $destination_storage = $entity_type_manager->getStorage('dataset_destination');
  /** @var \Drupal\data_pipelines\Entity\DestinationInterface $destination */
  foreach ($destination_storage->loadMultiple() as $destination) {
    if ($destination->getDestinationPluginId() === 'elasticsearch') {
      $data_set_destination = $destination;
      break;
    }
  }
  if (!$data_set_destination) {
    return t('No dataset destination found with elasticsearch plugin ID. Please create one and manually update all the data pipelines.');
  }
  $status_map = [
    'pending_validation' => DatasetInterface::STATUS_PENDING_VALIDATION,
    'pending_index' => DatasetInterface::STATUS_PENDING_PROCESSING,
    'indexed' => DatasetInterface::STATUS_PROCESSED,
  ];

  $dataset_storage = $entity_type_manager->getStorage('data_pipelines');
  /** @var \Drupal\data_pipelines\Entity\DatasetInterface $dataset */
  foreach ($dataset_storage->loadMultiple() as $dataset) {
    if (empty($dataset->getDestinations())) {
      // Update the status field values as per new mapping.
      $dataset->status->value = $status_map[$dataset->status->value];
      // Set the 'destinations' field value.
      $dataset->set('destinations', [$data_set_destination])
        ->save();
    }
  }
  return t('All the data pipelines are updated to use :label  dataset destination.', [
    ':label' => $data_set_destination->label(),
  ]);
}

/**
 * Update all the existing csv datasets with Comma as a default.
 */
function data_pipelines_post_update_set_delimiter(&$sandbox = NULL) {
  $storage = \Drupal::entityTypeManager()->getStorage('data_pipelines');
  $iteration_size = (int) Settings::get('entity_update_batch_size', 50);
  if (empty($sandbox['progress'])) {
    $sandbox['progress'] = 0;
    $sandbox['max'] = (int) $storage->getQuery()
      ->condition('source', 'csv:%', 'LIKE')
      ->accessCheck(FALSE)
      ->count()
      ->execute();
  }
  $start = $sandbox['progress'];
  $end = min($sandbox['max'], $start + $iteration_size);
  $ids = $storage->getQuery()
    ->condition('source', 'csv:%', 'LIKE')
    ->sort('id')
    ->accessCheck(FALSE)
    ->range($start, $end)
    ->execute();
  /** @var \Drupal\data_pipelines\Entity\DatasetInterface[] $datasets */
  $datasets = $storage->loadMultiple(array_values($ids));
  foreach ($datasets as $dataset) {
    $dataset->set('csv_delimiter', 'comma');
    $dataset->save();
  }
  if ($sandbox['max'] > 0 && $end < $sandbox['max']) {
    $sandbox['progress'] = $end;
    $sandbox['#finished'] = ($end - 1) / $sandbox['max'];
  }
  else {
    $sandbox['#finished'] = 1;
  }
}