influxdb-toolkit

Python package for querying InfluxDB v1 (InfluxQL) and v2 (Flux) with one consistent API.

Install

python -m pip install influxdb-toolkit

For local repo usage:

python -m pip install -e .

1-Minute Query (Auto-detect v1/v2)

from datetime import UTC, datetime, timedelta
from influxdb_toolkit import InfluxDBClientFactory

config = {
    "url": "http://localhost:8086",
    "token": "my-token",
    "org": "my-org",
    "bucket": "my-bucket",
}

with InfluxDBClientFactory.get_client(config=config) as client:
    df = client.get_timeseries(
        measurement="temperature",
        fields=["value"],
        start=datetime.now(UTC) - timedelta(hours=24),
        end=datetime.now(UTC),
        interval="5m",
        aggregation="mean",
        timezone="UTC",
    )
    print(df.head())

Copy/Paste Templates

V1 only

Use this when your InfluxDB v1 endpoint is exposed via HTTPS on port 443.

from datetime import UTC, datetime, timedelta
from influxdb_toolkit import InfluxDBClientFactory

config = {
    "host": "localhost",
    "port": 443,
    "username": "user",
    "password": "pass",
    "database": "mydb",
    "ssl": True,
}

with InfluxDBClientFactory.get_client(version=1, config=config) as client:
    df = client.get_timeseries(
        measurement="temperature",
        fields=["value"],
        start=datetime.now(UTC) - timedelta(hours=24),
        end=datetime.now(UTC),
        tags={"site": "A1"},
        interval="5m",
        aggregation="mean",
    )
    print(df.head())

V1 behind a Chronograf proxy

Use this when InfluxDB v1 is not directly reachable but is exposed through a Chronograf reverse proxy (common in hosted or containerised setups).

from datetime import UTC, datetime, timedelta
from influxdb_toolkit import InfluxDBClientFactory
from influxdb_toolkit.v1.chronograf_proxy import ChronografProxyClient

proxy = ChronografProxyClient(
    base_url="https://your-chronograf-host.example.com",
    username="user",
    password="pass",
    database="mydb",
    source_id="1",       # find via GET /chronograf/v1/sources
    verify_ssl=False,    # set True for a trusted cert
)

with InfluxDBClientFactory.get_client(version=1, config=proxy.toolkit_config, client=proxy) as client:
    df = client.get_timeseries(
        measurement="temperature",
        fields=["value"],
        start=datetime.now(UTC) - timedelta(hours=24),
        end=datetime.now(UTC),
        interval="5m",
        aggregation="mean",
    )
    print(df.head())

V2 only

from datetime import UTC, datetime, timedelta
from influxdb_toolkit import InfluxDBClientFactory

config = {
    "url": "http://localhost:8086",
    "token": "my-token",
    "org": "my-org",
    "bucket": "my-bucket",
}

with InfluxDBClientFactory.get_client(version=2, config=config) as client:
    df = client.get_timeseries(
        measurement="temperature",
        fields=["value"],
        start=datetime.now(UTC) - timedelta(hours=24),
        end=datetime.now(UTC),
        tags={"site": "A1"},
        interval="5m",
        aggregation="mean",
    )
    print(df.head())

Auto-detect (single script for mixed environments)

from datetime import UTC, datetime, timedelta
from influxdb_toolkit import InfluxDBClientFactory

# Use either v1 keys OR v2 keys.
config = {
    "url": "http://localhost:8086",
    "token": "my-token",
    "org": "my-org",
    "bucket": "my-bucket",
}

with InfluxDBClientFactory.get_client(config=config) as client:
    df = client.get_multiple_timeseries(
        queries=[
            {"measurement": "temperature", "fields": ["value"], "tags": {"site": "A1"}},
            {"measurement": "pressure", "fields": ["value"], "tags": {"site": "A1"}},
        ],
        start=datetime.now(UTC) - timedelta(hours=24),
        end=datetime.now(UTC),
        interval="10m",
        aggregation="mean",
    )
    print(df.head())

Common Query Tasks

Query one metric

df = client.get_timeseries(
    measurement="temperature",
    fields=["value"],
    start=start,
    end=end,
    tags={"site": "A1"},
    interval="5m",
    aggregation="mean",
)

Query multiple metrics together

df = client.get_multiple_timeseries(
    queries=[
        {"measurement": "temperature", "fields": ["value"], "tags": {"site": "A1"}},
        {"measurement": "pressure", "fields": ["value"], "tags": {"site": "A1"}},
    ],
    start=start,
    end=end,
    interval="10m",
    aggregation="mean",
)

Run raw query text

V1 InfluxQL:

df = client.query_raw(
    'SELECT mean("value") FROM "temperature" WHERE time >= now() - 24h GROUP BY time(5m)',
    timezone="UTC",
)

V2 Flux:

flux = '''
from(bucket: "my-bucket")
  |> range(start: -24h)
  |> filter(fn: (r) => r._measurement == "temperature")
  |> filter(fn: (r) => r._field == "value")
  |> aggregateWindow(every: 5m, fn: mean)
'''
df = client.query_raw(flux, timezone="UTC")

Explore schema before querying

measurements = client.list_measurements()
tag_keys = client.get_tags("temperature")
tag_values = client.get_tag_values("temperature", "site")
fields = client.get_fields("temperature")

List all series (measurement + tag combinations)

# All series across all measurements
df_series = client.get_all_series()
print(df_series)
#   measurement  device_id  sensor  site
#   temperature  dev_01     temp    A1
#   temperature  dev_01     hum     A1
#   humidity     dev_02     rh      B2

# Series for a single measurement
df_series = client.get_series(measurement="temperature")

Get actual fields per series (non-null detection)

# Which fields actually have data for each tag combination?
df_fields = client.get_fields_per_series(
    ignore_fields={"rssi_dBm_abs", "snr_dB_abs"},  # optional
)
print(df_fields)
#   measurement  device_id  sensor  fields                    n_fields
#   temperature  dev_01     temp    value, raw                2
#   humidity     dev_02     rh      rh                        1

Query multiple series with short column names

df = client.get_multiple_timeseries(
    queries=[
        {"measurement": "m", "fields": ["value"], "tags": {"sensor": "temp"}},
        {"measurement": "m", "fields": ["value"], "tags": {"sensor": "hum"}},
    ],
    start=start, end=end,
    short_names=True,  # columns: "value", "value_2" instead of full prefix
)

Get full database schema as DataFrame

df_schema = client.get_schema()
print(df_schema)
#   measurement    field     type          tags
#   temperature    value    float  site, sensor
#   humidity       rh       float  site, floor

# Filter by measurement
df_schema[df_schema["measurement"] == "temperature"]

# Filter by field type
df_schema[df_schema["type"] == "float"]

# List all unique measurements
df_schema["measurement"].unique()

Full Query API

• get_timeseries
• get_multiple_timeseries
• query_raw
• get_results_from_qry (alias to query_raw)
• list_measurements
• get_tags
• get_tag_values
• get_fields
• get_series
• get_all_series
• get_fields_per_series
• get_schema
• list_databases (v1)
• list_buckets (v2)

Config Rules

Factory version detection:

• v1 keys: host, database, username/password (or user/pwd)
• v2 keys: url, token, org

Environment variables are supported via .env.example.

Safety

Write/delete/admin operations are disabled by default.

Enable only if needed:

set INFLUXDB_ALLOW_WRITE=true

Developer Docs

If you are maintaining/extending the package:

• docs/usage_setup.md
• docs/architecture_concept.md
• CONTRIBUTING.md
• RELEASE.md

License

MIT. See LICENSE.