influxdb_service.py

import os
from dotenv import load_dotenv
from utils.fluxQueryBuilder import FluxQueryBuilder
from utils.influx import InfluxDBHelper

load_dotenv()

_bucket=os.getenv("INFLUXDB_BUCKET")
_org=os.getenv("INFLUXDB_ORG")

client = InfluxDBHelper(
    url=os.getenv("INFLUXDB_URL"),
    token=os.getenv("INFLUXDB_TOKEN"),
    org=_org,
    bucket=_bucket,
)

builder = (
    FluxQueryBuilder()
    .bucket(_bucket)
    .time_range("-30d", "now()")
    .filter_measurement("sensor_data")
    .filter_fields("co2", "rh", "temp") # with "or" we should implement "and" too 
    .mean()
)

"""
Some query examples
from(bucket: "co2-test")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "sensor_data")
  |> filter(fn: (r) => r["_field"] == "co2" or r["_field"] == "humidity" or r["_field"] == "temperature") 


from(bucket: "co2-test")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "sensor_data")
  |> filter(fn: (r) => r["_field"] == "co2" or r["_field"] == "humidity" or r["_field"] == "temperature")

from(bucket: "example-bucket")
    |> range(start: -1h)
    |> filter(fn: (r) => r._measurement == "cpu" and r._field == "usage_system" and r.cpu == "cpu-total")
    |> window(every: 5m) //every five min
    |> mean() //gives the average back

from(bucket: "co2-dev")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "sensor_data")
  |> filter(fn: (r) => r["room"] == "1/210")
  |> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)
  |> yield(name: "mean")

"""

flux_query = builder.build()

print(flux_query)

tables = client.query_api.query(org= _org, query= flux_query)

for table in tables:
    for record in table.records:
        print(record)