Class: VectorAmp::IngestionResource

Inherits:
Object
  • Object
show all
Defined in:
lib/vector_amp/ingestion.rb

Overview

Ingestion API resource for sources, jobs, and direct file uploads.

Instance Method Summary collapse

Constructor Details

#initialize(transport) ⇒ IngestionResource

Parameters:

  • transport (#request) —

    API transport.



13
14
15
 
# File 'lib/vector_amp/ingestion.rb', line 13

def initialize(transport)
  @transport = transport
end

Instance Method Details

#cancel_job(job_id) ⇒ Hash

Cancel an ingestion job.

Parameters:

  • job_id (String) —

    job id.

Returns:

  • (Hash) —

    cancel response.



299
300
301
 
# File 'lib/vector_amp/ingestion.rb', line 299

def cancel_job(job_id)
  @transport.request(:delete, "/ingestion/jobs/#{job_id}/cancel")
end

#cleanup_unused_sources ⇒ Hash

Delete all unused (unreferenced) sources.

Returns:

  • (Hash) —

    cleanup response.



51
52
53
 
# File 'lib/vector_amp/ingestion.rb', line 51

def cleanup_unused_sources
  @transport.request(:post, "/ingestion/sources/cleanup")
end

#complete_upload(source_id, job_id:, file_ids:) ⇒ Hash

Complete a file upload job after files have been PUT to presigned URLs.

Parameters:

  • source_id (String) —

    file-upload source id.

  • job_id (String) —

    upload job id from #init_upload.

  • file_ids (Array<String>) —

    file ids from #init_upload.

Returns:

  • (Hash) —

    upload completion/job response.



350
351
352
 
# File 'lib/vector_amp/ingestion.rb', line 350

def complete_upload(source_id, job_id:, file_ids:)
  @transport.request(:post, "/ingestion/sources/#{source_id}/upload/complete", body: { job_id: job_id, file_ids: file_ids })
end

#create(source = nil, **options) ⇒ Hash

Alias for #create_source.

Returns:

  • (Hash) —

    created source response.



91
92
93
 
# File 'lib/vector_amp/ingestion.rb', line 91

def create(source = nil, **options)
  create_source(source, **options)
end

#create_confluence(cloud_id: nil, base_url: nil, name: nil, auth_mode: "basic", username: nil, api_token: nil, spaces: nil, include_attachments: false, connection_id: nil, description: nil, metadata: nil, **config) ⇒ Hash

Create a Confluence source.

Parameters:

  • cloud_id (String, nil) (defaults to: nil) —

    Atlassian OAuth cloud/site id; required unless base_url is given.

  • base_url (String, nil) (defaults to: nil) —

    Confluence base URL, e.g. https://company.atlassian.net.

  • name (String, nil) (defaults to: nil) —

    defaults to confluence-<space>/confluence-<host>/confluence-source.

  • auth_mode (String) (defaults to: "basic") —

    basic (default) or oauth.

  • username (String, nil) (defaults to: nil) —

    username for basic auth.

  • api_token (String, nil) (defaults to: nil) —

    API token for basic auth.

  • spaces (String, Array<String>, nil) (defaults to: nil) —

    space keys to ingest; empty means all accessible.

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

    include page attachments; defaults to false.

  • connection_id (String, nil) (defaults to: nil) —

    optional managed connection id used in place of inline credentials.

  • description (String, nil) (defaults to: nil) —

    optional description.

  • metadata (Hash, nil) (defaults to: nil) —

    optional metadata.

  • config (Hash) —

    additional Confluence-source config forwarded to the API.

Returns:

  • (Hash) —

    created source response.



209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
 
# File 'lib/vector_amp/ingestion.rb', line 209

def create_confluence(cloud_id: nil, base_url: nil, name: nil, auth_mode: "basic", username: nil, api_token: nil,
                      spaces: nil, include_attachments: false, connection_id: nil, description: nil, metadata: nil, **config)
  create_source(ConfluenceSource.new(
    cloud_id: cloud_id,
    base_url: base_url,
    name: name,
    auth_mode: auth_mode,
    username: username,
    api_token: api_token,
    spaces: spaces,
    include_attachments: include_attachments,
    connection_id: connection_id,
    description: description,
    metadata: metadata,
    **config
  ))
end

#create_file_upload(name: nil, description: nil, metadata: nil, storage_provider: "s3", sync_mode: "full", **config) ⇒ Hash

Create a file-upload source.

Parameters:

  • name (String, nil) (defaults to: nil) —

    defaults to timestamped ruby-sdk-file-upload-YYYYmmddHHMMSS.

  • description (String, nil) (defaults to: nil) —

    optional description.

  • metadata (Hash, nil) (defaults to: nil) —

    optional metadata.

  • storage_provider (String) (defaults to: "s3") —

    storage backend; defaults to s3.

  • sync_mode (String) (defaults to: "full") —

    sync strategy; defaults to full.

  • config (Hash) —

    additional file-upload config forwarded to the API.

Returns:

  • (Hash) —

    created source response.



235
236
237
238
239
240
241
242
243
244
 
# File 'lib/vector_amp/ingestion.rb', line 235

def create_file_upload(name: nil, description: nil, metadata: nil, storage_provider: "s3", sync_mode: "full", **config)
  create_source(FileUploadSource.new(
    name: name,
    description: description,
    metadata: metadata,
    storage_provider: storage_provider,
    sync_mode: sync_mode,
    **config
  ))
end

#create_gcs(bucket:, name: nil, prefix: nil, connection_id: nil, description: nil, metadata: nil, **config) ⇒ Hash

Create a Google Cloud Storage source.

Parameters:

  • bucket (String) —

    required GCS bucket name.

  • name (String, nil) (defaults to: nil) —

    defaults to gcs-<bucket> or gcs-<bucket>-<prefix>.

  • prefix (String, nil) (defaults to: nil) —

    optional object prefix.

  • connection_id (String, nil) (defaults to: nil) —

    optional managed connection id used in place of inline credentials.

  • description (String, nil) (defaults to: nil) —

    optional description.

  • metadata (Hash, nil) (defaults to: nil) —

    optional metadata.

  • config (Hash) —

    additional GCS-source config forwarded to the API.

Returns:

  • (Hash) —

    created source response.



140
141
142
143
144
145
146
147
148
149
150
 
# File 'lib/vector_amp/ingestion.rb', line 140

def create_gcs(bucket:, name: nil, prefix: nil, connection_id: nil, description: nil, metadata: nil, **config)
  create_source(GCSSource.new(
    bucket: bucket,
    name: name,
    prefix: prefix,
    connection_id: connection_id,
    description: description,
    metadata: metadata,
    **config
  ))
end

#create_google_drive(name: nil, folder_ids: nil, file_ids: nil, auth_mode: nil, service_account_json: nil, oauth_credentials: nil, connection_id: nil, description: nil, metadata: nil, **config) ⇒ Hash

Create a Google Drive source.

Parameters:

  • name (String, nil) (defaults to: nil) —

    defaults to google-drive-<first id> or google-drive-source.

  • folder_ids (String, Array<String>, nil) (defaults to: nil) —

    folder ids to ingest; required if file_ids is empty.

  • file_ids (String, Array<String>, nil) (defaults to: nil) —

    file ids to ingest; required if folder_ids is empty.

  • auth_mode (String, nil) (defaults to: nil) —

    auth strategy (service_account, oauth); omitted when nil.

  • service_account_json (Hash, String, nil) (defaults to: nil) —

    service-account credentials for service_account auth.

  • oauth_credentials (Hash, nil) (defaults to: nil) —

    OAuth credentials for oauth auth.

  • connection_id (String, nil) (defaults to: nil) —

    optional managed connection id used in place of inline credentials.

  • description (String, nil) (defaults to: nil) —

    optional description.

  • metadata (Hash, nil) (defaults to: nil) —

    optional metadata.

  • config (Hash) —

    additional Google Drive-source config forwarded to the API.

Returns:

  • (Hash) —

    created source response.



179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
 
# File 'lib/vector_amp/ingestion.rb', line 179

def create_google_drive(name: nil, folder_ids: nil, file_ids: nil, auth_mode: nil, service_account_json: nil,
                        oauth_credentials: nil, connection_id: nil, description: nil, metadata: nil, **config)
  create_source(GoogleDriveSource.new(
    name: name,
    folder_ids: folder_ids,
    file_ids: file_ids,
    auth_mode: auth_mode,
    service_account_json: ,
    oauth_credentials: oauth_credentials,
    connection_id: connection_id,
    description: description,
    metadata: metadata,
    **config
  ))
end

#create_jira(cloud_id:, name: nil, access_token: nil, project_keys: nil, jql: nil, include_comments: true, connection_id: nil, description: nil, metadata: nil, **config) ⇒ Object 



152
153
154
155
156
157
158
159
160
161
162
163
164
165
 
# File 'lib/vector_amp/ingestion.rb', line 152

def create_jira(cloud_id:, name: nil, access_token: nil, project_keys: nil, jql: nil, include_comments: true, connection_id: nil, description: nil, metadata: nil, **config)
  create_source(JiraSource.new(
    cloud_id: cloud_id,
    name: name,
    access_token: access_token,
    project_keys: project_keys,
    jql: jql,
    include_comments: include_comments,
    connection_id: connection_id,
    description: description,
    metadata: metadata,
    **config
  ))
end

#create_s3(bucket:, name: nil, prefix: nil, description: nil, metadata: nil, **config) ⇒ Hash

Create an S3 source.

Parameters:

  • bucket (String) —

    required S3 bucket name.

  • name (String, nil) (defaults to: nil) —

    defaults to s3-<bucket> or s3-<bucket>-<prefix>.

  • prefix (String, nil) (defaults to: nil) —

    optional object prefix.

  • description (String, nil) (defaults to: nil) —

    optional description.

  • metadata (Hash, nil) (defaults to: nil) —

    optional metadata.

  • config (Hash) —

    additional S3-source config forwarded to the API.

Returns:

  • (Hash) —

    created source response.



120
121
122
123
124
125
126
127
128
129
 
# File 'lib/vector_amp/ingestion.rb', line 120

def create_s3(bucket:, name: nil, prefix: nil, description: nil, metadata: nil, **config)
  create_source(S3Source.new(
    name: name,
    bucket: bucket,
    prefix: prefix,
    description: description,
    metadata: metadata,
    **config
  ))
end

#create_source(source = nil, source_type: nil, name: nil, config: nil, description: nil, metadata: nil) ⇒ Hash

Create an ingestion source from a Source object/hash or explicit options.

Parameters:

  • source (Source, Hash, nil) (defaults to: nil) —

    optional source object/hash; when supplied, option fields are ignored.

  • source_type (String, Symbol, nil) (defaults to: nil) —

    source type (s3, web, gdrive, or file_upload).

  • name (String, nil) (defaults to: nil) —

    source name; defaults by source type when omitted.

  • config (Hash, nil) (defaults to: nil) —

    source-specific config.

  • description (String, nil) (defaults to: nil) —

    optional description.

  • metadata (Hash, nil) (defaults to: nil) —

    optional metadata.

Returns:

  • (Hash) —

    created source response.



78
79
80
81
82
83
84
85
86
87
 
# File 'lib/vector_amp/ingestion.rb', line 78

def create_source(source = nil, source_type: nil, name: nil, config: nil, description: nil, metadata: nil)
  body = source ? source_create_body(source) : source_create_body_from_options(
    source_type: source_type,
    name: name,
    description: description,
    config: config,
    metadata: metadata
  )
  @transport.request(:post, "/ingestion/sources", body: body)
end

#create_web(start_urls:, name: nil, description: nil, metadata: nil, **config) ⇒ Hash

Create a web source.

Parameters:

  • start_urls (String, Array<String>) —

    required seed URLs.

  • name (String, nil) (defaults to: nil) —

    defaults to web-<host> from the first URL, or web-source.

  • description (String, nil) (defaults to: nil) —

    optional description.

  • metadata (Hash, nil) (defaults to: nil) —

    optional metadata.

  • config (Hash) —

    additional web-source config forwarded to the API.

Returns:

  • (Hash) —

    created source response.



102
103
104
105
106
107
108
109
110
 
# File 'lib/vector_amp/ingestion.rb', line 102

def create_web(start_urls:, name: nil, description: nil, metadata: nil, **config)
  create_source(WebSource.new(
    name: name,
    start_urls: start_urls,
    description: description,
    metadata: metadata,
    **config
  ))
end

#delete_source(source_id, force: false) ⇒ Hash

Delete an ingestion source.

Parameters:

  • source_id (String) —

    source id.

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

    force deletion even if the source is still referenced; sends ?force=true.

Returns:

  • (Hash) —

    delete response.



36
37
38
39
 
# File 'lib/vector_amp/ingestion.rb', line 36

def delete_source(source_id, force: false)
  query = force ? { force: true } : nil
  @transport.request(:delete, "/ingestion/sources/#{source_id}", query: query)
end

#get_job(job_id) ⇒ Hash

Fetch an ingestion job.

Parameters:

  • job_id (String) —

    job id.

Returns:

  • (Hash) —

    job response.



271
272
273
 
# File 'lib/vector_amp/ingestion.rb', line 271

def get_job(job_id)
  @transport.request(:get, "/ingestion/jobs/#{job_id}")
end

#get_source(source_id) ⇒ Hash

Fetch an ingestion source.

Parameters:

  • source_id (String) —

    source id.

Returns:

  • (Hash) —

    source response.



28
29
30
 
# File 'lib/vector_amp/ingestion.rb', line 28

def get_source(source_id)
  @transport.request(:get, "/ingestion/sources/#{source_id}")
end

#ingest_files(dataset_id:, paths:, source_name: nil, description: nil, metadata: {}) ⇒ Hash

Upload local files by auto-creating a file_upload source, initializing presigned uploads, and completing the upload job.

Parameters:

  • dataset_id (String) —

    target dataset id; also added to source metadata.

  • paths (String, Array<String>) —

    local file paths to upload.

  • source_name (String, nil) (defaults to: nil) —

    optional source name; defaults to timestamped Ruby SDK file-upload name.

  • description (String, nil) (defaults to: nil) —

    optional source description.

  • metadata (Hash) (defaults to: {}) —

    optional source metadata merged with dataset_id.

Returns:

  • (Hash) —

    upload completion/job response.

Raises:

  • (ArgumentError) 


310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
 
# File 'lib/vector_amp/ingestion.rb', line 310

def ingest_files(dataset_id:, paths:, source_name: nil, description: nil, metadata: {})
  files = Array(paths).map { |path| Pathname(path) }
  raise ArgumentError, "paths must not be empty" if files.empty?

  source = create_file_upload(
    name: source_name,
    description: description,
    metadata: (metadata || {}).merge(dataset_id: dataset_id)
  )
  source_id = source.fetch("id") { source.fetch(:id) }

  init = init_upload(source_id, files)
  upload_files_to_presigned_urls(files, init.fetch("uploads"))
  job_id = init.fetch("job_id")
  response = complete_upload(source_id, job_id: job_id, file_ids: init.fetch("uploads").map { |upload| upload.fetch("file_id") })
  response["job_id"] ||= job_id if response.is_a?(Hash)
  response
end

#init_upload(source_id, files) ⇒ Hash

Initialize presigned uploads for source files.

Parameters:

  • source_id (String) —

    file-upload source id.

  • files (Array<String, Pathname>) —

    local files.

Returns:

  • (Hash) —

    init response containing uploads and job_id.



333
334
335
336
337
338
339
340
341
342
343
 
# File 'lib/vector_amp/ingestion.rb', line 333

def init_upload(source_id, files)
  payload = Array(files).map do |file|
    path = Pathname(file)
    {
      name: path.to_s,
      size_bytes: path.size,
      content_type: content_type_for(path)
    }
  end
  @transport.request(:post, "/ingestion/sources/#{source_id}/upload/init", body: { files: payload })
end

#job_files(job_id) ⇒ Hash

List files attached to an ingestion job.

Parameters:

  • job_id (String) —

    job id.

Returns:

  • (Hash) —

    files response.



285
286
287
 
# File 'lib/vector_amp/ingestion.rb', line 285

def job_files(job_id)
  @transport.request(:get, "/ingestion/jobs/#{job_id}/files")
end

#job_statistics(job_id) ⇒ Hash

Fetch ingestion job statistics.

Parameters:

  • job_id (String) —

    job id.

Returns:

  • (Hash) —

    statistics response.



292
293
294
 
# File 'lib/vector_amp/ingestion.rb', line 292

def job_statistics(job_id)
  @transport.request(:get, "/ingestion/jobs/#{job_id}/statistics")
end

#list_jobs(dataset_id: nil, limit: 50, offset: 0) ⇒ Hash

List ingestion jobs.

Parameters:

  • dataset_id (String, nil) (defaults to: nil) —

    optional dataset filter.

  • limit (Integer) (defaults to: 50) —

    page size; defaults to 50.

  • offset (Integer) (defaults to: 0) —

    page offset; defaults to 0.

Returns:

  • (Hash) —

    response envelope from the API.



264
265
266
 
# File 'lib/vector_amp/ingestion.rb', line 264

def list_jobs(dataset_id: nil, limit: 50, offset: 0)
  @transport.request(:get, "/ingestion/jobs", query: Utils.compact_hash(dataset_id: dataset_id, limit: limit, offset: offset))
end

#list_sources(limit: 50, offset: 0) ⇒ Hash

List ingestion sources.

Parameters:

  • limit (Integer) (defaults to: 50) —

    page size; defaults to 50.

  • offset (Integer) (defaults to: 0) —

    page offset; defaults to 0.

Returns:

  • (Hash) —

    response envelope from the API.



21
22
23
 
# File 'lib/vector_amp/ingestion.rb', line 21

def list_sources(limit: 50, offset: 0)
  @transport.request(:get, "/ingestion/sources", query: { limit: limit, offset: offset })
end

#list_unused_sources(limit: 50, offset: 0) ⇒ Hash

List sources that are not referenced by any job, schedule, or dataset.

Parameters:

  • limit (Integer) (defaults to: 50) —

    page size; defaults to 50.

  • offset (Integer) (defaults to: 0) —

    page offset; defaults to 0.

Returns:

  • (Hash) —

    response envelope from the API.



45
46
47
 
# File 'lib/vector_amp/ingestion.rb', line 45

def list_unused_sources(limit: 50, offset: 0)
  @transport.request(:get, "/ingestion/sources/unused", query: { limit: limit, offset: offset })
end

#retry_job(job_id) ⇒ Hash

Retry an eligible failed or cancelled ingestion job as a fresh full rerun.

Parameters:

  • job_id (String) —

    original job id.

Returns:

  • (Hash) —

    newly queued retry job response.



278
279
280
 
# File 'lib/vector_amp/ingestion.rb', line 278

def retry_job(job_id)
  @transport.request(:post, "/ingestion/jobs/#{job_id}/retry")
end

#source_references(source_id) ⇒ Hash

List references (jobs, schedules, datasets) that use a source.

Parameters:

  • source_id (String) —

    source id.

Returns:

  • (Hash) —

    references response.



58
59
60
 
# File 'lib/vector_amp/ingestion.rb', line 58

def source_references(source_id)
  @transport.request(:get, "/ingestion/sources/#{source_id}/references")
end

#start_job(source_id:, dataset_id:, pipeline_id: nil) ⇒ Hash

Start an ingestion job for a source and dataset.

Parameters:

  • source_id (String) —

    source id.

  • dataset_id (String) —

    target dataset id.

  • pipeline_id (String, nil) (defaults to: nil) —

    optional pipeline id.

Returns:

  • (Hash) —

    ingestion job response.



251
252
253
254
255
256
257
 
# File 'lib/vector_amp/ingestion.rb', line 251

def start_job(source_id:, dataset_id:, pipeline_id: nil)
  @transport.request(:post, "/ingestion/jobs", body: Utils.compact_hash(
    source_id: source_id,
    dataset_id: dataset_id,
    pipeline_id: pipeline_id
  ))
end

#validate_source(source_type:, config:) ⇒ Hash

Validate a source type and config without creating a source.

Parameters:

  • source_type (String, Symbol) —

    source type to validate.

  • config (Hash) —

    source-specific config to validate.

Returns:

  • (Hash) —

    validation response.



66
67
68
 
# File 'lib/vector_amp/ingestion.rb', line 66

def validate_source(source_type:, config:)
  @transport.request(:post, "/ingestion/sources/validate", body: { source_type: source_type, config: config })
end