Class: Altertable::Lakehouse::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/altertable/lakehouse/client.rb,
sig/altertable/lakehouse/client.rbs

Constant Summary collapse

DEFAULT_BASE_URL =

Returns:

  • (String) 
"https://api.altertable.ai"
DEFAULT_TIMEOUT =

Returns:

  • (Integer) 
10
DEFAULT_OPEN_TIMEOUT =

Returns:

  • (Integer) 
5

Instance Method Summary collapse

Constructor Details

#initialize(username: nil, password: nil, basic_auth_token: nil, base_url: nil, timeout: nil, open_timeout: nil, user_agent: nil, adapter: nil, headers: {}) ⇒ Client

Returns a new instance of Client.



15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/altertable/lakehouse/client.rb', line 15

def initialize(username: nil, password: nil, basic_auth_token: nil, base_url: nil, timeout: nil, open_timeout: nil, user_agent: nil, adapter: nil, headers: {})
  # 1. Try passed basic_auth_token
  # 2. Try passed username/password
  # 3. Try ENV["ALTERTABLE_BASIC_AUTH_TOKEN"]
  # 4. Try ENV["ALTERTABLE_USERNAME"] / ENV["ALTERTABLE_PASSWORD"]

  if basic_auth_token
    @auth_header = "Basic #{basic_auth_token}"
  elsif username && password
    @auth_header = "Basic #{Base64.strict_encode64("#{username}:#{password}")}"
  elsif (env_token = ENV["ALTERTABLE_BASIC_AUTH_TOKEN"])
    @auth_header = "Basic #{env_token}"
  elsif (env_user = ENV["ALTERTABLE_USERNAME"]) && (env_pass = ENV["ALTERTABLE_PASSWORD"])
    @auth_header = "Basic #{Base64.strict_encode64("#{env_user}:#{env_pass}")}"
  else
    raise ConfigurationError, "Authentication credentials required (username/password or basic_auth_token)"
  end

  @base_url = base_url || DEFAULT_BASE_URL
  @timeout = timeout || DEFAULT_TIMEOUT
  @open_timeout = open_timeout || DEFAULT_OPEN_TIMEOUT
  @user_agent = user_agent ? "AltertableRuby/#{VERSION} #{user_agent}" : "AltertableRuby/#{VERSION}"
  
  default_headers = {
    "Authorization" => @auth_header,
    "User-Agent" => @user_agent
  }

  @adapter = select_adapter(adapter, base_url: @base_url, timeout: @timeout, open_timeout: @open_timeout, headers: default_headers.merge(headers))
end

Instance Method Details

#append(catalog:, schema:, table:, payload:, sync: nil, headers: {}) ⇒ Object

POST /append



47
48
49
50
51
52
53
 
# File 'lib/altertable/lakehouse/client.rb', line 47

def append(catalog:, schema:, table:, payload:, sync: nil, headers: {})
  params = { catalog: catalog, schema: schema, table: table }
  params[:sync] = sync unless sync.nil?
  req = Models::AppendRequest.new(payload)
  resp = request(:post, "/append", body: req.to_h, query: params, headers: headers)
  Models::AppendResponse.from_h(resp)
end

#autocomplete(statement:, catalog: nil, schema: nil, session_id: nil, max_suggestions: nil, headers: {}) ⇒ Object

POST /autocomplete



145
146
147
148
149
150
151
152
153
154
155
 
# File 'lib/altertable/lakehouse/client.rb', line 145

def autocomplete(statement:, catalog: nil, schema: nil, session_id: nil, max_suggestions: nil, headers: {})
  req = Models::AutocompleteRequest.new(
    statement: statement,
    catalog: catalog,
    schema: schema,
    session_id: session_id,
    max_suggestions: max_suggestions
  )
  resp = request(:post, "/autocomplete", body: req.to_h, headers: headers)
  Models::AutocompleteResponse.from_h(resp)
end

#cancel_query(query_id, session_id:, headers: {}) ⇒ Models::CancelQueryResponse

DELETE /query/:query_id

Parameters:

  • query_id (String)
  • session_id: (String)
  • headers: (::Hash[String, String]) (defaults to: {})

Returns:



127
128
129
130
 
# File 'lib/altertable/lakehouse/client.rb', line 127

def cancel_query(query_id, session_id:, headers: {})
  resp = request(:delete, "/query/#{query_id}", query: { session_id: session_id }, headers: headers)
  Models::CancelQueryResponse.from_h(resp)
end

#explain(statement:, catalog: nil, schema: nil, session_id: nil, include_plan: nil, headers: {}) ⇒ Object

POST /explain



158
159
160
161
162
163
164
165
166
167
168
 
# File 'lib/altertable/lakehouse/client.rb', line 158

def explain(statement:, catalog: nil, schema: nil, session_id: nil, include_plan: nil, headers: {})
  req = Models::ExplainRequest.new(
    statement: statement,
    catalog: catalog,
    schema: schema,
    session_id: session_id,
    include_plan: include_plan
  )
  resp = request(:post, "/explain", body: req.to_h, headers: headers)
  Models::ExplainResponse.from_h(resp)
end

#get_query(query_id, headers: {}) ⇒ Models::QueryLogResponse

GET /query/:query_id

Parameters:

  • query_id (String)
  • headers: (::Hash[String, String]) (defaults to: {})

Returns:



121
122
123
124
 
# File 'lib/altertable/lakehouse/client.rb', line 121

def get_query(query_id, headers: {})
  resp = request(:get, "/query/#{query_id}", headers: headers)
  Models::QueryLogResponse.from_h(resp)
end

#get_task(task_id, headers: {}) ⇒ Models::TaskResponse

GET /tasks/:task_id

Parameters:

  • task_id (String)
  • headers: (::Hash[String, String]) (defaults to: {})

Returns:



56
57
58
59
 
# File 'lib/altertable/lakehouse/client.rb', line 56

def get_task(task_id, headers: {})
  resp = request(:get, "/tasks/#{task_id}", headers: headers)
  Models::TaskResponse.from_h(resp)
end

#handle_response(resp) ⇒ Object

Parameters:

Returns:

  • (Object) 


259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
 
# File 'lib/altertable/lakehouse/client.rb', line 259

def handle_response(resp)
  case resp.status
  when 200..299
    return nil if resp.body.nil? || resp.body.empty?
    begin
      JSON.parse(resp.body)
    rescue JSON::ParserError
      # For non-JSON responses
      resp.body
    end
  when 400
    raise BadRequestError, "Bad Request: #{resp.body}"
  when 401
    raise AuthError, "Unauthorized"
  when 404
    raise ApiError, "Not Found: #{resp.headers}" # Url not avail in struct easily
  else
    raise ApiError, "API Error #{resp.status}: #{resp.body}"
  end
end

#handle_stream_response(resp, buffer, yielder) ⇒ void

This method returns an undefined value.

Parameters:



222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
 
# File 'lib/altertable/lakehouse/client.rb', line 222

def handle_stream_response(resp, buffer, yielder)
  case resp.status
  when 400
    raise BadRequestError, "Bad Request: #{buffer.strip}"
  when 401
    raise AuthError, "Unauthorized"
  when 200..299
    # Parse the accumulated NDJSON buffer line by line
    # Buffer might be partial? 
    # In streaming, the block is called.
    # Here we are processing after the stream is done?
    # Wait, QueryResult expects the stream to be processed as it comes?
    # The previous implementation used an Enumerator that yielded as data came in.
    # Here, @adapter.post blocks until done?
    # If @adapter.post blocks, we only get the buffer at the end.
    # To stream truly, @adapter.post needs to yield to the block, which yields to yielder?
    
    # Re-implementing streaming logic:
    # The enumerator in `query` wraps the call. 
    # When `query` returns QueryResult, it hasn't run the request yet.
    # Enumerator logic is inside.
    
    buffer.each_line do |line|
      line = line.strip
      next if line.empty?
      begin
        yielder << JSON.parse(line)
      rescue JSON::ParserError
        # Partial line?
        # For now assume full lines or handle buffering properly
      end
    end
  else
    raise ApiError, "API Error #{resp.status}: #{buffer.strip}"
  end
end

#query(statement:, headers: {}, **options) ⇒ QueryResult

POST /query (streamed)

Parameters:

  • statement: (String)
  • headers: (::Hash[String, String]) (defaults to: {})
  • options (Object)

Returns:



62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/altertable/lakehouse/client.rb', line 62

def query(statement:, headers: {}, **options)
  req_body = Models::QueryRequest.new(statement: statement, **options).to_h.to_json

  enum = Enumerator.new do |yielder|
    buffer = ""
    
    # Use adapter's stream capability
    resp = @adapter.post("/query", body: req_body, headers: json_headers(headers)) do |chunk, _|
      buffer << chunk
    end

    handle_stream_response(resp, buffer, yielder)
  end

  QueryResult.new(enum)
end

#query_all(statement:, headers: {}, **options) ⇒ ::Hash[Symbol, untyped]

POST /query (accumulated)

Parameters:

  • statement: (String)
  • headers: (::Hash[String, String]) (defaults to: {})
  • options (Object)

Returns:

  • (::Hash[Symbol, untyped]) 


80
81
82
83
84
85
86
87
88
 
# File 'lib/altertable/lakehouse/client.rb', line 80

def query_all(statement:, headers: {}, **options)
  result = query(statement: statement, headers: headers, **options)
  rows = result.to_a # Accumulate
  {
    metadata: result.metadata,
    columns: result.columns,
    rows: rows
  }
end

#request(method, path, body: nil, query: nil, headers: {}) ⇒ Object 



199
200
201
202
203
204
205
206
207
 
# File 'lib/altertable/lakehouse/client.rb', line 199

def request(method, path, body: nil, query: nil, headers: {})
  resp = @adapter.send(
    method, path,
    body: encode_request_body(body),
    params: query || {},
    headers: body.nil? ? headers : json_headers(headers)
  )
  handle_response(resp)
end

#select_adapter(name, options) ⇒ Object

Parameters:

  • name (Symbol, nil)
  • options (Object)

Returns:

  • (Object) 


172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
 
# File 'lib/altertable/lakehouse/client.rb', line 172

def select_adapter(name, options)
  case name
  when :faraday
    Adapters::FaradayAdapter.new(**options)
  when :httpx
    Adapters::HttpxAdapter.new(**options)
  when :net_http
    Adapters::NetHttpAdapter.new(**options)
  else
    # Auto-detect
    if defined?(Faraday) || try_require("faraday")
      Adapters::FaradayAdapter.new(**options)
    elsif defined?(HTTPX) || try_require("httpx")
      Adapters::HttpxAdapter.new(**options)
    else
      Adapters::NetHttpAdapter.new(**options)
    end
  end
end

#try_require(gem_name) ⇒ Boolean

Parameters:

  • gem_name (String)

Returns:

  • (Boolean) 


192
193
194
195
196
197
 
# File 'lib/altertable/lakehouse/client.rb', line 192

def try_require(gem_name)
  require gem_name
  true
rescue LoadError
  false
end

#upload(catalog:, schema:, table:, mode:, file_io:, headers: {}) ⇒ Object

POST /upload



91
92
93
94
95
96
97
98
99
100
101
102
103
 
# File 'lib/altertable/lakehouse/client.rb', line 91

def upload(catalog:, schema:, table:, mode:, file_io:, headers: {})
  params = {
    catalog: catalog,
    schema: schema,
    table: table,
    mode: mode
  }

  body = file_io.respond_to?(:read) ? file_io.read : file_io

  resp = @adapter.post("/upload", body: body, params: params, headers: headers)
  handle_response(resp)
end

#upsert(catalog:, schema:, table:, primary_key:, file_io:, headers: {}) ⇒ Object

POST /upsert



106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/altertable/lakehouse/client.rb', line 106

def upsert(catalog:, schema:, table:, primary_key:, file_io:, headers: {})
  params = {
    catalog: catalog,
    schema: schema,
    table: table,
    primary_key: primary_key
  }

  body = file_io.respond_to?(:read) ? file_io.read : file_io

  resp = @adapter.post("/upsert", body: body, params: params, headers: headers)
  handle_response(resp)
end

#validate(statement:, catalog: nil, schema: nil, session_id: nil, headers: {}) ⇒ Object

POST /validate



133
134
135
136
137
138
139
140
141
142
 
# File 'lib/altertable/lakehouse/client.rb', line 133

def validate(statement:, catalog: nil, schema: nil, session_id: nil, headers: {})
  req = Models::ValidateRequest.new(
    statement: statement,
    catalog: catalog,
    schema: schema,
    session_id: session_id
  )
  resp = request(:post, "/validate", body: req.to_h, headers: headers)
  Models::ValidateResponse.from_h(resp)
end