grafane 2.1.0

Opinionated InfluxDB client

Grafane

A very opinionated InfluxDB client inspired by Grafana's query builder.

Setup

Installation

pip install grafane        # InfluxDB v2 (default)
pip install grafane[v1]    # Add InfluxDB v1 support

Or with poetry:

poetry add grafane             # InfluxDB v2 (default)
poetry add grafane --extras v1 # Add InfluxDB v1 support

Configuration

Grafane supports multi-database setups with a Django-style configuration system.

Recommended: Create a settings module in your project:

# myproject/settings.py
INFLUXDB_SETTINGS = {
    'default': {
        'version': 2,
        'url': 'http://localhost:8086',
        'token': 'my-api-token',
        'org': 'my-org',
        'bucket': 'my-bucket',
        'metrics': [],
    },
}

Then configure Grafane:

import grafane
grafane.configure('myproject.settings')

Alternatively, set the environment variable:

export GRAFANE_SETTINGS_MODULE=myproject.settings

Environment Variables (Default Configuration)

If no settings module is configured, Grafane uses these environment variables for a single default InfluxDB v2 database:

Variable	Default	Description
INFLUXDB_V2_URL	http://localhost:8086	InfluxDB v2 URL
INFLUXDB_V2_TOKEN	my-super-secret-token	API token
INFLUXDB_V2_ORG	my-org	Organization
INFLUXDB_V2_BUCKET	metrics	Bucket name
TESTING	0	If set, appends -testing to metric names

For multi-database setups, use a settings module instead (see Configuration above).

Quick Start

import grafane

# Configure with your settings module (optional if using env vars)
grafane.configure('myproject.settings')

# Create a client for a metric
c = grafane.Grafane(metric='temperature')

# Write
c.report({'value': 23.5}, {'room': 'living'})

# Read
results = c.select(fields='value').filter_by('room', '=', 'living').execute_query()

Multi-Database Setup

Configure multiple InfluxDB buckets in your settings module:

# myproject/settings.py
INFLUXDB_SETTINGS = {
    'default': {
        'version': 2,
        'url': 'http://localhost:8086',
        'token': 'my-api-token',
        'org': 'my-org',
        'bucket': 'metrics',
        'metrics': [],  # Empty = fallback for unmatched metrics
    },
    'analytics': {
        'version': 2,
        'url': 'http://analytics.example.com:8086',
        'token': 'analytics-token',
        'org': 'my-org',
        'bucket': 'analytics',
        'metrics': ['page_views', 'sessions', 'events'],
    },
}

v2 Configuration Keys

Key	Type	Required	Description
version	int	Yes	Set to 2 for InfluxDB v2
url	str	Yes	InfluxDB v2 URL
token	str	Yes	API token
org	str	Yes	Organization
bucket	str	Yes	Bucket name
metrics	list	No	Metrics routed to this bucket (empty = fallback)

Legacy InfluxDB v1 Support

Note: InfluxDB v1 requires the v1 extra: pip install grafane[v1]

INFLUXDB_SETTINGS = {
    'default': {
        'host': 'localhost',
        'port': 8086,
        'database': 'metrics',
        'username': 'admin',
        'password': 'secret',
        'metrics': [],
    },
}

v1 Configuration Keys:

Key	Type	Required	Description
host	str	Yes	InfluxDB host
port	int	Yes	InfluxDB port
database	str	Yes	Database name
username	str	Yes	Username
password	str	Yes	Password
metrics	list	No	Metrics routed to this database (empty = fallback)

Mixed v1/v2 Setup

You can configure both v1 and v2 databases in the same settings (requires pip install grafane[v1]):

INFLUXDB_SETTINGS = {
    'legacy': {
        'host': 'localhost',
        'port': 8086,
        'database': 'metrics_v1',
        'username': 'admin',
        'password': 'secret',
        'metrics': ['cpu', 'memory'],
    },
    'modern': {
        'version': 2,
        'url': 'http://localhost:8086',
        'token': 'my-token',
        'org': 'my-org',
        'bucket': 'metrics_v2',
        'metrics': ['events', 'traces'],
    },
}

Metric Routing

Grafane automatically routes metrics to the correct database based on the metrics list:

from grafane import Grafane

# Routes to 'analytics' (has 'page_views' in metrics list)
c = Grafane('page_views')

# Routes to 'monitoring' (has 'cpu' in metrics list)
c = Grafane('cpu')

# Routes to 'default' (fallback - empty metrics list)
c = Grafane('unknown_metric')

# Explicit database selection (bypasses routing)
c = Grafane('any_metric', db='analytics')

Routing rules:

1. If db= parameter is provided, use that database
2. If metric is in exactly one database's metrics list, use that database
3. If metric is in multiple databases' metrics lists, raises error (use db= to resolve)
4. If metric is not found, use the fallback database (database with empty metrics list)
5. If no fallback exists, raises error

Write

report()

c.report(fields, tags, timestamp=False)

report_points()

c.report_points([
    {'fields': {'value': 1.2}, 'tags': {'tag1': 'a'}},
    {'fields': {'value': 1.8}, 'tags': {'tag1': 'b'}},
])

If no timestamp is provided, defaults to datetime.now(pytz.utc).

Read

Chainable Query API

All query methods return self and can be chained:

results = (
    c.select(fields=['value', 'value2'], aggregation='mean')
    .filter_by('tag1', '=', 'value1')
    .time_block('1h')
    .fill_with('none')
    .execute_query()
)

select()

# Single field
c.select(fields='value')

# Multiple fields
c.select(fields=['value', 'value2'])

# With aggregation
c.select(fields='value', aggregation='sum')

# Multiple fields with different aggregations
c.select(fields=['value', 'value2'], aggregation=['sum', 'mean'])

filter_by()

Filter by tag with an operator:

c.filter_by('tag1', '=', 'value1')

Supported operators: =, !=, <, >, <=, >=, =~ (regex)

filter_value_in()

Match tags against multiple values:

c.filter_value_in('tag1', ['value1', 'value2'])

filter_time_range()

from datetime import datetime

time_range = (datetime(2024, 1, 1), datetime(2024, 1, 31))
c.filter_time_range(time_range)

Accepts tuple or list of datetime objects. Order doesn't matter.

time_block()

Group results by time intervals:

c.select(fields='value', aggregation='mean').time_block('1h')

fill_with()

Fill empty time blocks:

c.fill_with('none')

Options: none, null, 0, previous, linear

group_by()

Group by tag values (requires aggregation):

c.select(fields='value', aggregation='sum').group_by('tag1')

Iteration, Length, and Boolean

# Iterate directly
for row in c.select(fields='value'):
    print(row)

# Check result count
count = len(c.select(fields='value'))

# Boolean check
if c.select(fields='value').filter_by('tag1', '=', 'x'):
    print("Has results")

Drop Measurement

c = Grafane(metric='test')
c.drop_measurement()

Development

Docker Setup

Start services:

ahoy docker up

Services:

• metrics (InfluxDB 1.8): http://localhost:8086
• grafana: http://localhost:3000 (passwordless, admin access)

Jupyter Notebooks

ahoy notebooks run

Notebooks are stored in .notebooks/ directory.

Environment Files

Copy .env.copy to .env:

cp .env.copy .env

Contents:

INFLUXDB_DATA_ENGINE=tsm1
INFLUXDB_DB=metrics
INFLUXDB_USER=admin
INFLUXDB_PASSWORD=admin123