data_pipelines-1.x-dev/src/Entity/Dataset.php
src/Entity/Dataset.php
<?php
declare(strict_types=1);
namespace Drupal\data_pipelines\Entity;
use Drupal\Core\Entity\ContentEntityBase;
use Drupal\Core\Entity\EntityPublishedTrait;
use Drupal\Core\Entity\EntityStorageInterface;
use Drupal\Core\Entity\EntityTypeInterface;
use Drupal\Core\Field\BaseFieldDefinition;
use Drupal\Core\Field\FieldDefinitionInterface;
use Drupal\Core\Field\FieldStorageDefinitionInterface;
use Drupal\Core\Queue\QueueInterface;
use Drupal\Core\StringTranslation\TranslatableMarkup;
use Drupal\data_pipelines\DatasetPipelineInterface;
use Drupal\data_pipelines\DatasetPipelinePluginManager;
use Drupal\data_pipelines\Destination\DatasetDestinationPluginInterface;
use Drupal\data_pipelines\Destination\ProcessingOperation;
use Drupal\data_pipelines\Destination\ProcessingOperationEnum;
use Drupal\data_pipelines\Form\DatasetBatchContext;
use Drupal\data_pipelines\Source\DatasetSourceInterface;
use Drupal\data_pipelines\Source\DatasetSourcePluginManager;
use Drupal\data_pipelines\TransformValidDataIterator;
/**
* Defines a content entity for datasets.
*
* The bundles of this content entity are defined by the data_pipelines_source
* plugin manager.
*
* @ContentEntityType(
* id = "data_pipelines",
* label = @Translation("Dataset"),
* label_collection = @Translation("Datasets"),
* label_singular = @Translation("dataset"),
* label_plural = @Translation("datasets"),
* label_count = @PluralTranslation(
* singular = "@count dataset",
* plural = "@count datasets",
* ),
* bundle_label = @Translation("Dataset source"),
* admin_permission = "administer data_pipelines",
* base_table = "data_pipelines",
* translatable = FALSE,
* bundle_plugin_type = "data_pipelines_source",
* entity_keys = {
* "id" = "id",
* "bundle" = "source",
* "uuid" = "uuid",
* "label" = "name",
* "published" = "published",
* },
* handlers = {
* "storage" = \Drupal\data_pipelines\EntityHandlers\DatasetStorage::class,
* "list_builder" = \Drupal\data_pipelines\EntityHandlers\DatasetListBuilder::class,
* "views_data" = \Drupal\views\EntityViewsData::class,
* "route_provider" = {
* "html" = \Drupal\data_pipelines\EntityHandlers\DatasetRouteProvider::class,
* },
* "view_builder" = \Drupal\Core\Entity\EntityViewBuilder::class,
* "access" = \Drupal\data_pipelines\EntityHandlers\DatasetAccessHandler::class,
* "form" = {
* "add" = \Drupal\data_pipelines\Form\DatasetForm::class,
* "edit" = \Drupal\data_pipelines\Form\DatasetForm::class,
* "delete" = \Drupal\Core\Entity\ContentEntityDeleteForm::class,
* "process" = \Drupal\data_pipelines\Form\DatasetProcessForm::class,
* },
* },
* collection_permission = "data_pipelines list",
* links = {
* "collection" = "/admin/content/datasets",
* "edit-form" = "/admin/content/datasets/manage/{data_pipelines}/edit",
* "canonical" = "/admin/content/datasets/manage/{data_pipelines}",
* "add-page" = "/admin/content/datasets/add",
* "add-form" = "/admin/content/datasets/add/{source}",
* "delete-form" = "/data_pipelines/{data_pipelines}/delete",
* "process-form" = "/data_pipelines/{data_pipelines}/process",
* }
* )
*/
class Dataset extends ContentEntityBase implements DatasetInterface {
use EntityPublishedTrait;
/**
* {@inheritdoc}
*/
public static function baseFieldDefinitions(EntityTypeInterface $entity_type): array {
$fields = parent::baseFieldDefinitions($entity_type);
$fields['name'] = BaseFieldDefinition::create('string')
->setLabel(new TranslatableMarkup('Name'))
->setRequired(TRUE)
->setSetting('max_length', 255)
->setDisplayOptions('view', [
'label' => 'hidden',
'type' => 'string',
'weight' => -5,
])
->setDisplayOptions('form', [
'type' => 'string_textfield',
'weight' => -5,
])
->setDisplayConfigurable('form', TRUE);
// @todo change the form widget to 'machine_name' once
// https://www.drupal.org/project/drupal/issues/2685749 is fixed.
$fields['machine_name'] = BaseFieldDefinition::create('string')
->setLabel(new TranslatableMarkup('Machine Name'))
->setDescription(new TranslatableMarkup('The machine name used for indexing.'))
->setSetting('max_length', 128)
->setRequired(TRUE)
->addConstraint('UniqueField')
->addPropertyConstraints('value', ['Regex' => ['pattern' => '/^[a-z0-9_]+$/']]);
$fields['pipeline'] = BaseFieldDefinition::create('list_string')
->setLabel(new TranslatableMarkup('Pipeline'))
->setDescription(new TranslatableMarkup('The processing pipeline for the dataset.'))
->setRequired(TRUE)
->setSetting('allowed_values_function', self::class . '::getPipelines')
->setDisplayOptions('form', [
'type' => 'options_select',
]);
// Add the published field.
$fields += static::publishedBaseFieldDefinitions($entity_type);
$fields['published']
->setDescription(new TranslatableMarkup('Unpublishing the pipeline will remove all the destination data.'))
->setRevisionable(FALSE)
->setTranslatable(FALSE)
->setInitialValue(TRUE)
->setDisplayOptions('view', [
'type' => 'boolean',
'format' => 'yes-no',
'label' => 'above',
'weight' => -4,
])
->setDisplayOptions('form', [
'type' => 'options_select',
'weight' => 50,
]);
$fields['status'] = BaseFieldDefinition::create('list_string')
->setLabel(new TranslatableMarkup('Status'))
->setDefaultValue(self::STATUS_INITIAL)
->setSetting('allowed_values', [
self::STATUS_PENDING_VALIDATION => new TranslatableMarkup('Pending validation'),
self::STATUS_PENDING_PROCESSING => new TranslatableMarkup('Pending processing'),
self::STATUS_PROCESSED => new TranslatableMarkup('Processed'),
self::STATUS_INITIAL => new TranslatableMarkup('Initial state'),
])
->setDisplayConfigurable('view', TRUE)
->setDisplayOptions('view', [
'type' => 'list_default',
'label' => 'above',
'weight' => -4,
])
->setRequired(TRUE);
$fields['log'] = BaseFieldDefinition::create('string_long')
->setLabel(new TranslatableMarkup('Log'))
->setDescription(new TranslatableMarkup('Log entries.'))
->setDisplayConfigurable('view', TRUE)
->setDisplayOptions('view', [
'type' => 'basic_string',
'label' => 'above',
'weight' => 40,
])
->setCardinality(FieldStorageDefinitionInterface::CARDINALITY_UNLIMITED)
->setDisplayConfigurable('view', TRUE)
->setDisplayConfigurable('form', TRUE);
$fields['destinations'] = BaseFieldDefinition::create('entity_reference')
->setLabel(t('Destinations'))
->setDescription(t('The output destinations.'))
->setRequired(TRUE)
->setCardinality(FieldStorageDefinitionInterface::CARDINALITY_UNLIMITED)
->setSetting('target_type', 'dataset_destination')
->setDisplayOptions('form', [
'type' => 'options_select',
]);
$fields['batch_size'] = BaseFieldDefinition::create('list_integer')
->setLabel(t('Batch size'))
->setDescription(new TranslatableMarkup('The number of items to process in each batch operation. Datasets with only limited data in each row can select a large value here to increase performance. Datasets with a lot of data in each row should select a small value here to prevent out of memory issues.'))
->setSetting('allowed_values', [
10 => '10',
100 => '100',
1000 => '1,000',
10000 => '10,000',
100000 => '100,000',
])
->setRequired(TRUE)
->setInitialValue(1000)
->setDefaultValue(1000)
->setDisplayConfigurable('form', TRUE)
->setDisplayOptions('form', [
'type' => 'options_select',
'label' => 'above',
'weight' => 10,
]);
$fields['invalid_values'] = BaseFieldDefinition::create('list_string')
->setLabel(t('Invalid values handling'))
->setDescription(new TranslatableMarkup('Select the approach to be used for invalid values that were previously valid.'))
->setSetting('allowed_values', [
DatasetInterface::INVALID_RETAIN => new TranslatableMarkup('Retain records that were previously valid'),
DatasetInterface::INVALID_REMOVE => new TranslatableMarkup('Remove records that are now invalid and outdated'),
])
->setRequired(TRUE)
->setInitialValue(DatasetInterface::INVALID_RETAIN)
->setDefaultValue(DatasetInterface::INVALID_RETAIN)
->setDisplayConfigurable('form', TRUE)
->setDisplayOptions('form', [
'type' => 'options_select',
'label' => 'above',
'weight' => 10,
]);
return $fields;
}
/**
* An method to get the plugins.
*
* The plugins are keyed using the plugin Id and the value being the plugin
* label.
*
* @param \Drupal\Core\Field\FieldDefinitionInterface $field_definition
* The field definition.
* @param \Drupal\data_pipelines\Entity\Dataset $dataset
* The dataset entity.
*
* @return array
* The plugins.
*/
public static function getPipelines(FieldDefinitionInterface $field_definition, Dataset $dataset): array {
return array_map(function (array $definition) {
return $definition['label'];
}, \Drupal::service('plugin.manager.data_pipelines_pipeline')->getDefinitions());
}
/**
* Gets the source ID.
*
* @return string
* The source.
*/
protected function getSourceId(): string {
return $this->get('source')->value;
}
/**
* {@inheritdoc}
*/
public function getPipelineId(): string {
return $this->get('pipeline')->value;
}
/**
* {@inheritdoc}
*/
public function getPipeline(): DatasetPipelineInterface {
return \Drupal::service('plugin.manager.data_pipelines_pipeline')->createInstance($this->getPipelineId());
}
/**
* Gets the source plugin.
*
* @return \Drupal\data_pipelines\Source\DatasetSourceInterface
* Source plugin.
*/
protected function getSource(): DatasetSourceInterface {
return \Drupal::service('plugin.manager.data_pipelines_source')->createInstance($this->getSourceId());
}
/**
* {@inheritdoc}
*/
public function getDestinations(): array {
$destinations = [];
foreach ($this->get('destinations') as $destination) {
$destinations[] = $destination->entity ?? NULL;
}
return array_filter($destinations);
}
/**
* {@inheritdoc}
*/
public function queueProcessing(QueueInterface $queue, DatasetBatchContext $context): void {
if (!$this->isPublished()) {
return;
}
if ($context->getProcessed() === 0) {
$queue->deleteQueue();
}
$destinations = [];
$last_deltas = [];
$chunk_sizes = [];
$buffers = [];
foreach ($this->getDestinations() as $destination_entity) {
assert($destination_entity instanceof DestinationInterface);
$destination_id = $destination_entity->id();
// Keep track of destinations, so we can easily loop them by ID later.
$destinations[$destination_id] = $destination_entity;
// Get the last delta from the dataset.
$last_deltas[$destination_id] = $destination_entity->getDestinationPlugin()->getLastDelta($this);
// Store a keyed array of the chunk size each destination wants to store
// in each batch entry.
$chunk_sizes[$destination_id] = $destination_entity->getDestinationPlugin()->getProcessingChunkSize();
// Buffer dataset rows for each destination, so we can create the queue
// items with the appropriate number of rows (per the chunk size).
$buffers[$destination_id] = [];
if ($context->getProcessed() === 0) {
// If we're processing from the start, create a queue item for a 'begin'
// operation.
$queue->createItem(new ProcessingOperation(ProcessingOperationEnum::Begin, $destination_id));
}
}
$iterator = $this->getDataIterator($context->getProcessed());
$valid = 0;
// Iterate over each row in the dataset.
foreach ($iterator as $delta => $transformed) {
$valid++;
foreach ($destinations as $destination_id => $destination) {
assert($destination instanceof DestinationInterface);
// Push this item into the buffer for each destination. Be sure to
// retain the original delta as this may be significant, e.g. it might
// be used to generate a unique ID for each row.
$buffers[$destination_id][$delta] = $transformed;
if ($chunk_sizes[$destination_id] === DatasetDestinationPluginInterface::PROCESSING_CHUNK_SIZE_ALL) {
// This destination wants all rows at once (if possible) so we don't
// need to flush the buffer until we've processed as many records as
// we can.
continue;
}
if (count($buffers[$destination_id]) === $chunk_sizes[$destination_id]) {
// The number of records in the buffer for this destination matches
// the requested chunk size, create a queue item for the buffered
// records and this destination plugin.
$queue->createItem(new ProcessingOperation(ProcessingOperationEnum::ProcessChunk, $destination_id, $buffers[$destination_id]));
// Then flush the buffer.
$buffers[$destination_id] = [];
}
}
if ($valid === $context->getChunkSize()) {
break;
}
}
if ($valid < $context->getChunkSize()) {
$context->setFinished();
}
$invalid_deltas = $iterator->getInvalidDeltas();
$invalid = count($invalid_deltas);
$context->incrementValid($valid);
$context->incrementInvalid($invalid);
// We've processed all available records at this point, so we need to flush
// any buffered data.
foreach ($destinations as $destination_id => $destination) {
assert($destination instanceof DestinationInterface);
// There are buffered records for this destination.
if ($buffers[$destination_id]) {
// So we create a queue item for the buffered records and this
// destination.
$queue->createItem(new ProcessingOperation(ProcessingOperationEnum::ProcessChunk, $destination_id, $buffers[$destination_id]));
}
// Queue cleanup of any invalid items.
if ($this->getInvalidValuesHandling() === self::INVALID_REMOVE && $invalid_deltas) {
$queue->createItem(new ProcessingOperation(ProcessingOperationEnum::ProcessCleanup, $destination_id, $invalid_deltas));
}
// We've also reached the final record, so we queue an 'end' operation.
if ($context->isFinished()) {
// If the last delta is less than the current last delta,
// truncate upwards starting from the next delta.
if ($context->getLastDelta() < $last_deltas[$destination_id] && $this->getInvalidValuesHandling() === self::INVALID_REMOVE) {
$queue->createItem(new ProcessingOperation(ProcessingOperationEnum::ProcessCleanup, $destination_id, range($context->getLastDelta() + 1, $last_deltas[$destination_id])));
}
$queue->createItem(new ProcessingOperation(ProcessingOperationEnum::End, $destination_id));
}
}
}
/**
* {@inheritdoc}
*/
public function data(): \Generator {
@trigger_error(sprintf('Calling %s is deprecated in data_pipelines:1.0.0-alpha19 and removed in data_pipelines:2.0.0. Use ::getDataIterator instead. See https://www.drupal.org/project/data_pipelines/issues/3361740', __METHOD__), E_USER_DEPRECATED);
$source = $this->getSource();
$pipeline = $this->getPipeline();
foreach ($source->extractDataFromDataSet($this) as $data) {
yield $pipeline->transform($data);
}
}
/**
* {@inheritdoc}
*/
public function getDataIterator(int $seek_from = 0): TransformValidDataIterator {
$pipeline = $this->getPipeline();
$source = $this->getSource();
return new TransformValidDataIterator($source->extractDataFromDataSet($this), $this, $pipeline, $seek_from);
}
/**
* {@inheritdoc}
*/
public function resetLogs(bool $auto_save = TRUE): DatasetInterface {
$this->log = NULL;
$this->status = self::STATUS_INITIAL;
if ($auto_save) {
$this->save();
}
return $this;
}
/**
* {@inheritdoc}
*/
public function addLogMessage(string $message, bool $auto_save = TRUE): DatasetInterface {
$this->log[] = $message;
if ($auto_save) {
$this->save();
}
return $this;
}
/**
* {@inheritdoc}
*/
public function getStatusLabel(): TranslatableMarkup {
$map = $this->getFieldDefinition('status')->getSetting('allowed_values');
return $map[$this->get('status')->value] ?? new TranslatableMarkup('Unknown');
}
/**
* {@inheritdoc}
*/
public function getLogs(): array {
return array_column($this->get('log')->getValue(), 'value');
}
/**
* {@inheritdoc}
*/
public function setPendingValidation(): DatasetInterface {
$this->status = self::STATUS_PENDING_VALIDATION;
$this->save();
return $this;
}
/**
* {@inheritdoc}
*/
public function setPendingProcessing(): DatasetInterface {
$this->status = self::STATUS_PENDING_PROCESSING;
$this->save();
return $this;
}
/**
* {@inheritdoc}
*/
public function setPendingDeletion(): DatasetInterface {
$this->status = self::STATUS_PENDING_DELETION;
$this->save();
return $this;
}
/**
* {@inheritdoc}
*/
public function setProcessingComplete(): DatasetInterface {
$this->status->value = self::STATUS_PROCESSED;
$this->save();
return $this;
}
/**
* {@inheritdoc}
*/
public function isPendingValidation(): bool {
return $this->status->value === self::STATUS_PENDING_VALIDATION;
}
/**
* {@inheritdoc}
*/
public function isPendingProcessing(): bool {
return $this->status->value === self::STATUS_PENDING_PROCESSING;
}
/**
* {@inheritdoc}
*/
public function isPendingDeletion(): bool {
return $this->status->value === self::STATUS_PENDING_DELETION;
}
/**
* {@inheritdoc}
*/
public function isProcessed(): bool {
return $this->status->value === self::STATUS_PROCESSED;
}
/**
* {@inheritdoc}
*/
public function getMachineName(): ?string {
return $this->machine_name->value;
}
/**
* {@inheritdoc}
*/
public function getPipelineLabel() {
$manager = \Drupal::service('plugin.manager.data_pipelines_pipeline');
assert($manager instanceof DatasetPipelinePluginManager);
$plugin_id = $this->getPipelineId();
if ($manager->hasDefinition($plugin_id)) {
return $manager->getDefinition($plugin_id)['label'];
}
return $plugin_id;
}
/**
* {@inheritdoc}
*/
public function getSourceLabel() {
$manager = \Drupal::service('plugin.manager.data_pipelines_source');
assert($manager instanceof DatasetSourcePluginManager);
$plugin_id = $this->getSourceId();
return $manager->getDefinition($plugin_id)['label'];
}
/**
* {@inheritdoc}
*/
public static function loadByMachineName(string $machine_name) {
$entity_type_repository = \Drupal::service('entity_type.repository');
$entity_type_manager = \Drupal::entityTypeManager();
$storage = $entity_type_manager->getStorage($entity_type_repository->getEntityTypeFromClass(static::class));
// Build a query to fetch the entity IDs.
$entity_query = $storage->getQuery();
$entity_query->accessCheck(FALSE);
$entity_query->condition('machine_name', trim($machine_name));
$result = $entity_query->execute();
return $result ? $storage->load(reset($result)) : NULL;
}
/**
* {@inheritdoc}
*/
public function postSave(EntityStorageInterface $storage, $update = TRUE) {
parent::postSave($storage, $update);
if (!$this->isPublished()) {
foreach ($this->getDestinations() as $destination) {
$destination->getDestinationPlugin()->deleteDataSet($this, $destination);
}
}
\Drupal::service('plugin.manager.queue_worker')->clearCachedDefinitions();
}
/**
* {@inheritdoc}
*/
public static function postDelete(EntityStorageInterface $storage, array $entities) {
parent::postDelete($storage, $entities);
\Drupal::service('plugin.manager.queue_worker')->clearCachedDefinitions();
}
/**
* {@inheritdoc}
*/
public function hasValidation(): bool {
return $this->getPipeline()->hasValidation();
}
/**
* {@inheritdoc}
*/
public function getBatchSize(): int {
return (int) ($this->get('batch_size')->value ?? 1000);
}
/**
* {@inheritdoc}
*/
public function getProcessingQueueId(): string {
return DatasetInterface::QUEUE_NAME_PREFIX . $this->getMachineName();
}
/**
* {@inheritdoc}
*/
public function getInvalidValuesHandling(): string {
return $this->get('invalid_values')->value ?? self::INVALID_RETAIN;
}
}
