feat: add parameterized queries (#96)

* feat: add query params to model

* feat: add new resource types

* feat: add parameterized queries

Co-authored-by: Jakub Bednar <[email protected]>

Author: michaelahojna

CHANGELOG.md

@@ -1,5 +1,11 @@
 ## 2.2.0 [unreleased]
 
+### Features
+1. [#96](https://github.com/influxdata/influxdb-client-ruby/pull/96): Add support for Parameterized Queries
+
+### Documentation
+1. [#96](https://github.com/influxdata/influxdb-client-ruby/pull/96): Add Parameterized Queries example
+
 ## 2.1.0 [2021-10-22]
 
 ### Features

README.md

@@ -152,6 +152,33 @@ query_api.query_stream(query: query).each do |record|
 end
 ```
 
+#### Parameterized queries
+InfluxDB Cloud supports [Parameterized Queries](https://docs.influxdata.com/influxdb/cloud/query-data/parameterized-queries/)
+that let you dynamically change values in a query using the InfluxDB API. Parameterized queries make Flux queries more
+reusable and can also be used to help prevent injection attacks.
+
+InfluxDB Cloud inserts the params object into the Flux query as a Flux record named `params`. Use dot or bracket
+notation to access parameters in the `params` record in your Flux query. Parameterized Flux queries support only `int`
+, `float`, and `string` data types. To convert the supported data types into
+other [Flux basic data types, use Flux type conversion functions](https://docs.influxdata.com/influxdb/cloud/query-data/parameterized-queries/#supported-parameter-data-types).
+
+Parameterized query example:
+> :warning: Parameterized Queries are supported only in InfluxDB Cloud, currently there is no support in InfluxDB OSS.
+
+```ruby
+client = InfluxDB2::Client.new('https://localhost:8086', 'my-token',
+                              bucket: 'my-bucket',
+                              org: 'my-org')
+
+query = 'from(bucket: params.bucketParam) |> range(start: duration(v: params.startParam))'
+params = { 'bucketParam' => 'my-bucket', 'startParam' => '-1h' }
+
+query_api = client.create_query_api
+result = query_api.query(query: query, params: params)
+
+result[0].records.each { |record| puts "#{record.time} #{record.measurement}: #{record.field} #{record.value}" }
+```
+
 ### Writing data
 The [WriteApi](https://github.com/influxdata/influxdb-client-ruby/blob/master/lib/influxdb2/client/write_api.rb) supports synchronous and batching writes into InfluxDB 2.0. In default api uses synchronous write. To enable batching you can use WriteOption.
 

apis/lib/influxdb2/apis/generated/models/resource.rb

@@ -1,7 +1,7 @@
 =begin
-#Influx OSS API Service
+#InfluxDB OSS API Service
 
-#No description provided (generated by Openapi Generator https://github.com/openapitools/openapi-generator)
+#The InfluxDB v2 API provides a programmatic interface for all interactions with InfluxDB. Access the InfluxDB API using the `/api/v2/` endpoint. 
 
 The version of the OpenAPI document: 2.0.0
 
@@ -76,10 +76,6 @@ def self.openapi_types
     # List of attributes with nullable: true
     def self.openapi_nullable
       Set.new([
-        :'id',
-        :'name',
-        :'org_id',
-        :'org'
       ])
     end
 
@@ -134,15 +130,15 @@ def list_invalid_properties
     # @return true if the model is valid
     def valid?
       return false if @type.nil?
-      type_validator = EnumAttributeValidator.new('String', ["authorizations", "buckets", "dashboards", "orgs", "sources", "tasks", "telegrafs", "users", "variables", "scrapers", "secrets", "labels", "views", "documents", "notificationRules", "notificationEndpoints", "checks", "dbrp", "notebooks", "annotations"])
+      type_validator = EnumAttributeValidator.new('String', ["authorizations", "buckets", "dashboards", "orgs", "sources", "tasks", "telegrafs", "users", "variables", "scrapers", "secrets", "labels", "views", "documents", "notificationRules", "notificationEndpoints", "checks", "dbrp", "notebooks", "annotations", "remotes", "replications"])
       return false unless type_validator.valid?(@type)
       true
     end
 
     # Custom attribute writer method checking allowed values (enum).
     # @param [Object] type Object to be assigned
     def type=(type)
-      validator = EnumAttributeValidator.new('String', ["authorizations", "buckets", "dashboards", "orgs", "sources", "tasks", "telegrafs", "users", "variables", "scrapers", "secrets", "labels", "views", "documents", "notificationRules", "notificationEndpoints", "checks", "dbrp", "notebooks", "annotations"])
+      validator = EnumAttributeValidator.new('String', ["authorizations", "buckets", "dashboards", "orgs", "sources", "tasks", "telegrafs", "users", "variables", "scrapers", "secrets", "labels", "views", "documents", "notificationRules", "notificationEndpoints", "checks", "dbrp", "notebooks", "annotations", "remotes", "replications"])
       unless validator.valid?(type)
         fail ArgumentError, "invalid value for \"type\", must be one of #{validator.allowable_values}."
       end

examples/README.md

@@ -0,0 +1,13 @@
+# Examples
+
+## Writes
+
+
+## Queries
+- [influxdb_18_example.rb](influxdb_18_example.rb) - How to connect to InfluxDB 1.8
+- [parameterized_query.rb](parameterized_query.rb) - How to use parameterized Flux queries
+
+## Management API
+- [create_new_bucket.rb](create_new_bucket.rb) - How to create Buckets
+
+## Others

examples/parameterized_query.rb

@@ -0,0 +1,34 @@
+$LOAD_PATH.unshift File.expand_path('../lib', __dir__)
+require 'influxdb-client'
+
+# warning: Parameterized Queries are supported only in InfluxDB Cloud, currently there is no support in InfluxDB OSS.
+
+url = 'https://europe-west1-1.gcp.cloud2.influxdata.com'
+token = 'my-token'
+bucket = 'my-bucket'
+org = 'my-org'
+
+client = InfluxDB2::Client.new(url,
+                               token,
+                               bucket: bucket,
+                               org: org,
+                               precision: InfluxDB2::WritePrecision::NANOSECOND)
+
+puts '*** Write Points ***'
+
+write_api = client.create_write_api
+point = InfluxDB2::Point.new(name: 'weather')
+                        .add_tag('location', 'Praque')
+                        .add_field('temperature', 21)
+puts point.to_line_protocol
+write_api.write(data: point)
+
+puts '*** Query Points ***'
+
+query_api = client.create_query_api
+query = 'from(bucket: params.bucketParam) |> range(start: duration(v: params.startParam))'
+params = { 'bucketParam' => 'my-bucket', 'startParam' => '-1h' }
+result = query_api.query(query: query, params: params)
+result[0].records.each { |record| puts "#{record.time} #{record.measurement}: #{record.field} #{record.value}" }
+
+client.close!

lib/influxdb2/client/models/query.rb

@@ -1,7 +1,7 @@
 =begin
-#Influx OSS API Service
+#InfluxDB OSS API Service
 
-#No description provided (generated by Openapi Generator https://github.com/openapitools/openapi-generator)
+#The InfluxDB v2 API provides a programmatic interface for all interactions with InfluxDB. Access the InfluxDB API using the `/api/v2/` endpoint. 
 
 The version of the OpenAPI document: 2.0.0
 
@@ -24,6 +24,9 @@ class Query
     # The type of query. Must be \"flux\".
     attr_reader :type
 
+    # Enumeration of key/value pairs that respresent parameters to be injected into query (can only specify either this field or extern and not both) 
+    attr_accessor :params
+
     attr_accessor :dialect
 
     # Specifies the time that should be reported as \"now\" in the query. Default is the server's now time.
@@ -57,6 +60,7 @@ def self.attribute_map
         :'extern' => :'extern',
         :'query' => :'query',
         :'type' => :'type',
+        :'params' => :'params',
         :'dialect' => :'dialect',
         :'now' => :'now',
       }
@@ -68,6 +72,7 @@ def self.openapi_types
         :'extern' => :'File',
         :'query' => :'String',
         :'type' => :'String',
+        :'params' => :'Hash<String, Object>',
         :'dialect' => :'Dialect',
         :'now' => :'Time'
       }
@@ -106,6 +111,12 @@ def initialize(attributes = {})
         self.type = attributes[:'type']
       end
 
+      if attributes.key?(:'params')
+        if (value = attributes[:'params']).is_a?(Hash)
+          self.params = value
+        end
+      end
+
       if attributes.key?(:'dialect')
         self.dialect = attributes[:'dialect']
       end
@@ -153,6 +164,7 @@ def ==(o)
           extern == o.extern &&
           query == o.query &&
           type == o.type &&
+          params == o.params &&
           dialect == o.dialect &&
           now == o.now
     end
@@ -166,7 +178,7 @@ def eql?(o)
     # Calculates hash code according to all attributes.
     # @return [Integer] Hash code
     def hash
-      [extern, query, type, dialect, now].hash
+      [extern, query, type, params, dialect, now].hash
     end
 
     # Builds the object from hash

lib/influxdb2/client/query_api.rb

@@ -36,16 +36,18 @@ def initialize(options:)
 
     # @param [Object] query the flux query to execute. The data could be represent by [String], [Query]
     # @param [String] org specifies the source organization
+    # @param [Enumerable] params represent key/value pairs parameters to be injected into query
     # @return [String] result of query
-    def query_raw(query: nil, org: nil, dialect: DEFAULT_DIALECT)
-      _post_query(query: query, org: org, dialect: dialect).read_body
+    def query_raw(query: nil, org: nil, dialect: DEFAULT_DIALECT, params: nil)
+      _post_query(query: query, org: org, dialect: dialect, params: params).read_body
     end
 
     # @param [Object] query the flux query to execute. The data could be represent by [String], [Query]
     # @param [String] org specifies the source organization
+    # @param [Enumerable] params represent key/value pairs parameters to be injected into query
     # @return [Array] list of FluxTables which are matched the query
-    def query(query: nil, org: nil, dialect: DEFAULT_DIALECT)
-      response = query_raw(query: query, org: org, dialect: dialect)
+    def query(query: nil, org: nil, dialect: DEFAULT_DIALECT, params: nil)
+      response = query_raw(query: query, org: org, dialect: dialect, params: params)
       parser = InfluxDB2::FluxCsvParser.new(response)
 
       parser.parse
@@ -54,20 +56,21 @@ def query(query: nil, org: nil, dialect: DEFAULT_DIALECT)
 
     # @param [Object] query the flux query to execute. The data could be represent by [String], [Query]
     # @param [String] org specifies the source organization
+    # @param [Enumerable] params represent key/value pairs parameters to be injected into query
     # @return stream of Flux Records
-    def query_stream(query: nil, org: nil, dialect: DEFAULT_DIALECT)
-      response = _post_query(query: query, org: org, dialect: dialect)
+    def query_stream(query: nil, org: nil, dialect: DEFAULT_DIALECT, params: nil)
+      response = _post_query(query: query, org: org, dialect: dialect, params: params)
 
       InfluxDB2::FluxCsvParser.new(response, stream: true)
     end
 
     private
 
-    def _post_query(query: nil, org: nil, dialect: DEFAULT_DIALECT)
+    def _post_query(query: nil, org: nil, dialect: DEFAULT_DIALECT, params: nil)
       org_param = org || @options[:org]
       _check('org', org_param)
 
-      payload = _generate_payload(query, dialect)
+      payload = _generate_payload(query: query, dialect: dialect, params: params)
       return nil if payload.nil?
 
       uri = _parse_uri('/api/v2/query')
@@ -76,16 +79,17 @@ def _post_query(query: nil, org: nil, dialect: DEFAULT_DIALECT)
       _post_json(payload.to_body.to_json, uri)
     end
 
-    def _generate_payload(query, dialect)
+    def _generate_payload(query: nil, dialect: nil, params: nil)
       if query.nil?
         nil
       elsif query.is_a?(Query)
+        query.params = params unless params.nil?
         query
       elsif query.is_a?(String)
         if query.empty?
           nil
         else
-          Query.new(query: query, dialect: dialect, type: nil)
+          Query.new(query: query, dialect: dialect, type: nil, params: params)
         end
       end
     end

test/influxdb/query_api_test.rb

@@ -49,6 +49,27 @@ def test_query_raw
     assert_equal result, SUCCESS_DATA
   end
 
+  def test_parameterized_query_raw
+    body = '{"query":"from(bucket: params.bucketParam) |> range(start:  duration(v: params.startParam)) |> last()",' \
+'"params":{"bucketParam":"my-bucket","startParam":"1970-01-01T00:00:00.000000001Z"},' \
+'"dialect":{"header":true,"delimiter":",","annotations":["datatype","group","default"],' \
+'"commentPrefix":"#","dateTimeFormat":"RFC3339"}}'
+    stub_request(:post, 'http://localhost:8086/api/v2/query?org=my-org')
+      .with(body: body)
+      .to_return(body: SUCCESS_DATA)
+
+    client = InfluxDB2::Client.new('http://localhost:8086', 'my-token',
+                                   bucket: 'my-bucket',
+                                   org: 'my-org',
+                                   use_ssl: false)
+
+    query = 'from(bucket: params.bucketParam) |> range(start:  duration(v: params.startParam)) |> last()'
+    params = Hash['bucketParam' => 'my-bucket', 'startParam' => '1970-01-01T00:00:00.000000001Z']
+    result = client.create_query_api.query_raw(query: query, params: params)
+
+    assert_equal result, SUCCESS_DATA
+  end
+
   def test_query
     stub_request(:post, 'http://localhost:8086/api/v2/query?org=my-org')
       .to_return(body: SUCCESS_DATA)
@@ -73,6 +94,36 @@ def test_query
     assert_equal 'free', record1.field
   end
 
+  def test_parameterized_query
+    body = '{"query":"from(bucket: params.bucketParam) |> range(start:  duration(v: params.startParam)) |> last()",' \
+'"params":{"bucketParam":"my-bucket","startParam":"1970-01-01T00:00:00.000000001Z"},' \
+'"dialect":{"header":true,"delimiter":",","annotations":["datatype","group","default"],' \
+'"commentPrefix":"#","dateTimeFormat":"RFC3339"}}'
+    stub_request(:post, 'http://localhost:8086/api/v2/query?org=my-org')
+      .with(body: body)
+      .to_return(body: SUCCESS_DATA)
+
+    client = InfluxDB2::Client.new('http://localhost:8086', 'my-token',
+                                   bucket: 'my-bucket',
+                                   org: 'my-org',
+                                   use_ssl: false)
+
+    query = 'from(bucket: params.bucketParam) |> range(start:  duration(v: params.startParam)) |> last()'
+    params = Hash['bucketParam' => 'my-bucket', 'startParam' => '1970-01-01T00:00:00.000000001Z']
+
+    result = client.create_query_api.query(query: query, params: params)
+
+    assert_equal 1, result.length
+    assert_equal 4, result[0].records.length
+
+    record1 = result[0].records[0]
+
+    assert_equal Time.parse('1970-01-01T00:00:10Z').to_datetime.rfc3339(9), record1.time
+    assert_equal 'mem', record1.measurement
+    assert_equal 10, record1.value
+    assert_equal 'free', record1.field
+  end
+
   def test_headers
     stub_request(:post, 'http://localhost:8086/api/v2/query?org=my-org')
       .to_return(body: SUCCESS_DATA)