feat:add standard generate template of influxdb query statment

This commit is contained in:
douxu 2024-08-06 17:03:41 +08:00
parent 5e42854213
commit d0fe5f6982
3 changed files with 64 additions and 20 deletions

View File

@ -7,7 +7,7 @@ mongodb_database: "wave_record"
influxdb_host: "localhost"
influxdb_port: "8086"
influxdb_token: ""
influxdb_token: "lCuiQ316qlly3iFeoi1EUokPJ0XxW-5lnG-3rXsKaaZSjfuxO5EaZfFdrNGM7Zlrdk1PrN_7TOsM_SCu9Onyew=="
influxdb_org: "coslight"
influxdb_bucket: "wave_record"

View File

@ -21,12 +21,14 @@ func WriterPointIntoInfluxDB(ctx context.Context, writeAPI api.WriteAPIBlocking,
// QueryPointFromInfluxDB return the result of query from influxdb by special parameters
// Get query client
// queryAPI := client.QueryAPI(org)
func QueryPointFromInfluxDB(ctx context.Context, queryAPI api.QueryAPI, parameters InfluxDBPara, logger *zap.Logger) ([]interface{}, error) {
func QueryPointFromInfluxDB(ctx context.Context, queryAPI api.QueryAPI, logger *zap.Logger) ([]interface{}, error) {
dataList := make([]interface{}, 0)
// get query result
results, err := queryAPI.QueryWithParams(ctx, query, parameters)
// TODO 增加初始化示例
query := InitQueryByPara()
results, err := queryAPI.Query(ctx, query)
if err != nil {
logger.Error("query data by parameters failed", zap.Error(err), zap.String("para", parameters.String()))
logger.Error("query data by parameters failed", zap.Error(err))
return nil, err
}

View File

@ -1,23 +1,65 @@
// Package database define database operation functions
package database
import "fmt"
import (
"fmt"
"strings"
)
var query = `from(bucket:params.bucket)
|> range(start: duration(params.start))
|> filter(fn: (r) => r._measurement == parames.measurement)
|> filter(fn: (r) => r._field == params.field)
|> filter(fn: (r) => r._value > params.value)`
// InfluxDBPara define struct of influxdb query parameters
type InfluxDBPara struct {
Bucket string `json:"bucket"`
Measurement string `json:"measurement"`
Start string `json:"start"`
Field string `json:"field"`
Value float64 `json:"value"`
// QueryPara define struct of influxdb query parameters
type InfluxDBQueryPara struct {
Bucket string `json:"bucket"`
Measurement string `json:"measurement"`
Start string `json:"start"`
Stop string `json:"stop"`
Field string `json:"field"`
// Value float64 `json:"value"`
}
func (i InfluxDBPara) String() string {
return fmt.Sprintf("bucket:%s, measurement:%s, start:%s, field:%s, value:%f", i.Bucket, i.Measurement, i.Start, i.Field, i.Value)
type QueryPara interface {
apply(string) string
}
type BucketQueryPara string
func (b BucketQueryPara) apply(query string) string {
template := " from(bucket:\"{bucketId}\") "
template = strings.Replace(template, "{bucketId}", string(b), -1)
return query + template
}
func WithBucket(bucketID string) BucketQueryPara {
return BucketQueryPara(bucketID)
}
type MeasurementQueryPara string
func (m MeasurementQueryPara) apply(query string) string {
template := " |> filter(fn: (r) => r[\"_measurement\"] == \"{measurement}\") "
template = strings.Replace(template, "{measurement}", string(m), -1)
return query + template
}
func WithMeasurement(measurement string) MeasurementQueryPara {
return MeasurementQueryPara(measurement)
}
type StartTimeQueryPara string
func (s StartTimeQueryPara) apply(query string) string {
template := " |> range(start: {start}) "
template = strings.Replace(template, "{start}", string(s), -1)
return query + template
}
func WithStartTime(measurement string) StartTimeQueryPara {
return StartTimeQueryPara(measurement)
}
func InitQueryByPara(paras ...QueryPara) (queryCmd string) {
for _, para := range paras {
queryCmd = para.apply(queryCmd)
}
fmt.Printf("queryCmd:%v\n", queryCmd)
return
}