Class: OpenAI::Internal::VectorStoreFileUploader Private

Inherits:
Object
  • Object
show all
Defined in:
lib/openai/internal/vector_store_file_uploader.rb,
sig/openai/internal/vector_store_file_uploader.rbs

Overview

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

Uploads files for a vector store with bounded concurrency.

The caller enumerates and snapshots stream-backed inputs before workers start, so lazy and stateful Ruby enumerables are never advanced from worker threads. Originally streamed inputs are reopened as streams for files.create, retaining its non-retryable request semantics. Results retain input order.

API:

  • private

Constant Summary collapse

MAX_FILES_PER_BATCH =

This constant is part of a private API. You should avoid using this constant if possible, as it may be removed or be changed in the future.

The API limit for files attached in one vector store batch.

Returns:

API:

  • private

2_000

Instance Method Summary collapse

Constructor Details

#initialize(client:, max_concurrency:, request_options:) ⇒ VectorStoreFileUploader

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of VectorStoreFileUploader.

Parameters:

API:

  • private



29
30
31
32
33
34
35
36
37
 
# File 'lib/openai/internal/vector_store_file_uploader.rb', line 29

def initialize(client:, max_concurrency:, request_options:)
  unless max_concurrency.is_a?(Integer) && max_concurrency.positive?
    raise ArgumentError, "`max_concurrency` must be a positive integer"
  end

  @client = client
  @max_concurrency = max_concurrency
  @request_options = OpenAI::Internal::RequestOptionsScope.new(request_options)
end

Instance Method Details

#upload(files, max_files: MAX_FILES_PER_BATCH) ⇒ Array<OpenAI::Models::FileObject>

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Parameters:

  • (defaults to: MAX_FILES_PER_BATCH)

    Maximum number of inputs that may be uploaded.

Returns:

API:

  • private



44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/openai/internal/vector_store_file_uploader.rb', line 44

def upload(files, max_files: MAX_FILES_PER_BATCH)
  staged, temporary_files = stage(files, max_files: max_files)
  return [] if staged.empty?

  worker_count = [@max_concurrency, staged.length].min
  start_gate = Queue.new
  lock = Mutex.new
  uploaded = []
  state = {error: nil, next_index: 0, stopping: false}
  workers = []

  Thread.handle_interrupt(Exception => :never) do
    start_workers(start_gate, staged, worker_count, lock, uploaded, state, workers)
    Thread.handle_interrupt(Exception => :immediate) do
      start_gate.close
      workers.each(&:join)
    end

  ensure
    stop_workers(start_gate, workers, lock, state)
  end

  error = lock.synchronize { state[:error] }
  raise error unless error.nil?

  uploaded
ensure
  temporary_files&.each { remove_temporary_file(_1) }
end