Class: ElasticGraph::JSONIngestion::IngestionAdapter

Inherits:
Object
  • Object
show all
Defined in:
elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/ingestion_adapter.rb

Overview

Ingestion adapter for events in ElasticGraph's versioned JSON format: it validates events against the JSON schema identified by the event's json_schema_version, and prepares records using that version's view of the schema. Made available to the indexer by the IndexerExtension that SchemaDefinition::APIExtension registers.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(schema_artifacts:, logger:, configure_record_validator: nil) ⇒ IngestionAdapter

Returns a new instance of IngestionAdapter.

Parameters:

  • schema_artifacts (SchemaArtifacts::FromDisk) —

    the schema artifacts

  • logger (Logger) —

    the ElasticGraph logger

  • configure_record_validator (Proc, nil) (defaults to: nil) —

    optional callback to further configure the record validator



32
33
34
35
36
37
38
39
 
# File 'elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/ingestion_adapter.rb', line 32

def initialize(schema_artifacts:, logger:, configure_record_validator: nil)
  @envelope_validator = EnvelopeValidator.new(
    schema_artifacts: schema_artifacts,
    logger: logger,
    configure_record_validator: configure_record_validator
  )
  @record_preparer_factory = RecordPreparerFactory.new(schema_artifacts)
end

Instance Attribute Details

#envelope_validator ⇒ EnvelopeValidator (readonly)

Returns validates decoded JSON payloads before operations are built from them.

Returns:

  • (EnvelopeValidator) —

    validates decoded JSON payloads before operations are built from them



27
28
29
 
# File 'elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/ingestion_adapter.rb', line 27

def envelope_validator
  @envelope_validator
end

Instance Method Details

#validate_event(event, skip_record_validation: false) ⇒ ElasticGraph::Indexer::IngestionAdapter::ValidationResult

Validates the record of the given event and resolves the record preparer appropriate for the event's JSON schema version.

Parameters:

  • event (Hash<String, Object>) —

    an ElasticGraph indexing event

  • skip_record_validation (Boolean) (defaults to: false) —

    whether to skip record validation

Returns:

  • (ElasticGraph::Indexer::IngestionAdapter::ValidationResult) —

    the result of validating the event



47
48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/ingestion_adapter.rb', line 47

def validate_event(event, skip_record_validation: false)
  # Envelope validation has already confirmed that a version can be selected for this event.
  selected_json_schema_version = @envelope_validator.closest_available_json_schema_version(event.fetch(JSON_SCHEMA_VERSION_KEY)) # : ::Integer

  # The datastore includes `id` in search payloads only when it is part of the indexed record.
  record = event.fetch("record").merge("id" => event.fetch("id"))
  graphql_type_name = event.fetch("type")

  if !skip_record_validation && (error_message = @envelope_validator.validator(graphql_type_name, selected_json_schema_version).validate_with_error_message(record))
    return ValidationResult.invalid(validation_target: "#{graphql_type_name} record", message: error_message)
  end

  ValidationResult.valid(event.merge("record" => record), @record_preparer_factory.for_json_schema_version(selected_json_schema_version))
end