query weatherdata in influx

api/rest-api.go

@@ -40,9 +40,20 @@ func (api *weatherRestApi) handleRequests() *mux.Router {
 	router.HandleFunc("/random", api.randomWeatherHandler)
 	router.HandleFunc("/randomlist", api.randomWeatherListHandler)
 	router.HandleFunc("/addData", api.addDataHandler)
+	router.HandleFunc("/getData", api.getData)
 	return router
 }
 
+func (api *weatherRestApi) getData(w http.ResponseWriter, r *http.Request) {
+	w.Header().Add("content-type", "application/json")
+	data, err := api.weaterStorage.GetData()
+	if err != nil {
+		http.Error(w, "", http.StatusBadRequest)
+		return
+	}
+	json.NewEncoder(w).Encode(data)
+}
+
 func (api *weatherRestApi) randomWeatherHandler(w http.ResponseWriter, r *http.Request) {
 	datapoint := storage.NewRandomWeatherData("swablab")
 

storage/influxdb-storage.go

@@ -1,6 +1,10 @@
 package storage
 
 import (
+	"context"
+	"fmt"
+	"time"
+
 	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
 )
 
@@ -10,6 +14,7 @@ type influxStorage struct {
 	bucket       string
 	organization string
 	url          string
+	measurement  string
 	client       influxdb2.Client
 }
 
@@ -21,6 +26,7 @@ func NewInfluxStorage(token, bucket, organization, url string) (*influxStorage,
 	influx.organization = organization
 	influx.url = url
 	influx.client = influxdb2.NewClient(url, token)
+	influx.measurement = "data"
 	return influx, nil
 }
 
@@ -34,7 +40,7 @@ func (storage *influxStorage) Save(data WeatherData) error {
 		"humidity":    data.Humidity,
 		"pressure":    data.Pressure}
 
-	datapoint := influxdb2.NewPoint("new2",
+	datapoint := influxdb2.NewPoint(storage.measurement,
 		tags,
 		fields,
 		data.TimeStamp)
@@ -44,6 +50,65 @@ func (storage *influxStorage) Save(data WeatherData) error {
 	return nil
 }
 
+//GetData datapoints from InfluxDB
+func (storage *influxStorage) GetData() ([]*WeatherData, error) {
+
+	query := fmt.Sprintf("from(bucket:\"%v\")|> range(start: -40m, stop: -20m) |> filter(fn: (r) => r._measurement == \"data\" and r.location == \"Hamburg\")", storage.bucket)
+
+	res, err := storage.executeFluxQuery(query)
+	return res, err
+}
+
+func (storage *influxStorage) executeFluxQuery(query string) ([]*WeatherData, error) {
+
+	queryAPI := storage.client.QueryAPI(storage.organization)
+	result, err := queryAPI.Query(context.Background(), query)
+
+	if err != nil {
+		return nil, err
+	}
+
+	var queryResults []*WeatherData
+
+	for result.Next() {
+		if result.Err() != nil {
+			return nil, err
+		}
+		location := result.Record().ValueByKey("location").(string)
+		timestamp := result.Record().Time()
+
+		data, contained := containsWeatherData(queryResults, location, timestamp)
+
+		if result.Record().Field() == "temperature" {
+			data.Temperature = result.Record().Value().(float64)
+		}
+		if result.Record().Field() == "pressure" {
+			data.Pressure = result.Record().Value().(float64)
+		}
+		if result.Record().Field() == "humidity" {
+			data.Humidity = result.Record().Value().(float64)
+		}
+
+		if !contained {
+			data.Location = location
+			data.TimeStamp = timestamp
+			queryResults = append(queryResults, data)
+		}
+	}
+
+	return queryResults, nil
+}
+
+func containsWeatherData(weatherData []*WeatherData, location string, timestamp time.Time) (*WeatherData, bool) {
+	for _, val := range weatherData {
+		if val.Location == location && val.TimeStamp == timestamp {
+			return val, true
+		}
+	}
+	var newData WeatherData
+	return &newData, false
+}
+
 //Close InfluxDB connection
 func (storage *influxStorage) Close() error {
 	storage.client.Close()

storage/weather-data.go

@@ -8,6 +8,7 @@ import (
 //WeatherStorage interface for different storage-implementations of weather data
 type WeatherStorage interface {
 	Save(WeatherData) error
+	GetData() ([]*WeatherData, error)
 	Close() error
 }