Class: Lucid::Shopify::BulkRequest::Operation

Inherits:
Object
  • Object
show all
Defined in:
lib/lucid/shopify/bulk_request.rb

Instance Method Summary collapse

Instance Method Details

#call(delay: 1, http: Container[:http]) {|Enumerator<Hash>| ... } ⇒ Object

Wait for the operation to complete, then download the JSONL result data which is yielded as an Enumerator to the block. The data is streamed and parsed line by line to limit memory usage.

Parameters:

  • delay (Integer) (defaults to: 1) —

    delay between polling requests in seconds

  • http (HTTP::Client) (defaults to: Container[:http])

Yields:

  • (Enumerator<Hash>) —

    yields each parsed line of JSONL

Raises:

  • CanceledOperationError

  • ExpiredOperationError

  • FailedOperationError

  • ObsoleteOperationError



42
43
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
73
74
75
76
77
78
 
# File 'lib/lucid/shopify/bulk_request.rb', line 42

def call(delay: 1, http: Container[:http], &block)
  url = loop do
    status, url = poll

    case status
    when 'CANCELED'
      raise CanceledOperationError
    when 'EXPIRED'
      raise ExpiredOperationError
    when 'FAILED'
      raise FailedOperationError
    when 'COMPLETED'
      break url
    else
      sleep(delay)
    end
  end

  return if url.nil?

  # TODO: Verify signature?

  begin
    file = Tempfile.new(mode: 0600)
    body = http.get(url).body
    until (chunk = body.readpartial).nil?
      file.write(chunk)
    end
    file.rewind
    block.(Enumerator.new do |y|
      file.each_line { |line| y << JSON.parse(line) }
    end)
  ensure
    file.close
    file.unlink
  end
end

#cancel ⇒ Object

Cancel the bulk operation.

Raises:

  • ObsoleteOperationError

  • TimeoutError



84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/lucid/shopify/bulk_request.rb', line 84

def cancel
  begin
    client.post_graphql(credentials, <<~QUERY)
      mutation {
        bulkOperationCancel(id: "#{id}") {
          userErrors {
            field
            message
          }
        }
      }
    QUERY
  rescue Response::GraphQLClientError => e
    return if e.response.error_message?([
      /cannot be canceled when it is completed/,
    ])

    raise e
  end

  poll_until(['CANCELED', 'COMPLETED'])
end

#poll_until(statuses, timeout: 60) ⇒ Object

Poll until operation status is met.

Parameters:

  • statuses (Array<String>) —

    to terminate polling on

  • timeout (Integer) (defaults to: 60) —

    in seconds

Raises:

  • ObsoleteOperationError

  • TimeoutError



114
115
116
117
118
119
120
121
122
123
124
125
126
127
 
# File 'lib/lucid/shopify/bulk_request.rb', line 114

def poll_until(statuses, timeout: 60)
  Timeout.timeout(timeout) do
    loop do
      status, _ = poll

      break if statuses.any? { |expected_status| status == expected_status }
    end
  end
rescue Timeout::Error
  raise TimeoutError, 'exceeded %s seconds polling for status %s' % [
    timeout,
    statuses.join(', '),
  ]
end