InfluxDB with Python

InfluxDB provides an official Python client library for writing and querying data programmatically.

Setup

Install the client library:

pip install influxdb-client

---

Writing Data

Connection Setup

from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS

token = "<YOUR TOKEN HERE>"
org = "my-org"
bucket = "my-bucket"

client = InfluxDBClient(url="http://localhost:8086", token=token, org=org)
write_api = client.write_api(write_options=SYNCHRONOUS)

Replace <YOUR TOKEN HERE> with the API token you generated in the InfluxDB UI.

Writing with Line Protocol

Data can be written as a line protocol string. The format is:

measurement,tag_key=tag_value field_key=field_value

measurement = "temperature"
tags = "location=office"
fields = "current_temp=21.5"

data = f"{measurement},{tags} {fields}"
write_api.write(bucket=bucket, record=data)

Line Protocol Format

Pay attention to the formatting:

• Comma between measurement and tags (no space)
• Space before fields
• Multiple tags: tag1=val1,tag2=val2
• Multiple fields: field1=val1,field2=val2

Writing with the Point API

The Point class provides a more readable way to construct data points:

from influxdb_client import Point

point = (
    Point("temperature")
    .tag("location", "office")
    .field("current_temp", 21.5)
)

write_api.write(bucket=bucket, record=point)

Closing the Connection

Always close the client when you are done:

client.close()

---

Complete Write Example

This example writes a simulated temperature reading every 5 seconds:

import time
import random
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS

token = "<YOUR TOKEN HERE>"
org = "my-org"
bucket = "sensors"

client = InfluxDBClient(url="http://localhost:8086", token=token, org=org)
write_api = client.write_api(write_options=SYNCHRONOUS)

current_temp = 0.0

try:
    while True:
        # Simulate temperature change (+/- 0.5 degrees)
        change = random.uniform(-0.5, 0.5)
        current_temp += change

        point = (
            Point("temperature")
            .tag("location", "office")
            .field("current_temp", round(current_temp, 2))
        )

        write_api.write(bucket=bucket, record=point)
        print(f"Wrote: {current_temp:.2f} C")

        time.sleep(5)
except KeyboardInterrupt:
    print("Stopped.")
finally:
    client.close()

Run the script and check the InfluxDB Data Explorer to see the data appear.

---

Querying Data

Running a Flux Query

from influxdb_client import InfluxDBClient

token = "<YOUR TOKEN HERE>"
org = "my-org"

client = InfluxDBClient(url="http://localhost:8086", token=token, org=org)
query_api = client.query_api()

query = '''
from(bucket: "sensors")
  |> range(start: -1d)
  |> filter(fn: (r) => r["_measurement"] == "temperature")
  |> filter(fn: (r) => r["location"] == "office")
  |> filter(fn: (r) => r["_field"] == "current_temp")
'''

result = query_api.query(query, org=org)

The Flux query string is written as a multi-line Python string (triple quotes).

Processing Results

Query results are returned as a list of tables, where each table contains records.

for table in result:
    for record in table.records:
        print(f"Time: {record.get_time()}, "
              f"Field: {record.get_field()}, "
              f"Value: {record.get_value()}")

client.close()

Each record provides:

Method	Returns
record.get_time()	Timestamp of the data point
record.get_field()	Field name (e.g., "current_temp")
record.get_value()	Field value (e.g., 21.5)
record.get_measurement()	Measurement name (e.g., "temperature")

---

Complete Query Example

This example queries the last hour of temperature data and prints it:

from influxdb_client import InfluxDBClient

token = "<YOUR TOKEN HERE>"
org = "my-org"

client = InfluxDBClient(url="http://localhost:8086", token=token, org=org)
query_api = client.query_api()

query = '''
from(bucket: "sensors")
  |> range(start: -1h)
  |> filter(fn: (r) => r["_measurement"] == "temperature")
  |> filter(fn: (r) => r["_field"] == "current_temp")
'''

result = query_api.query(query, org=org)

for table in result:
    for record in table.records:
        print(f"{record.get_time()} | {record.get_value()} C")

client.close()

---

System Monitoring Example

This example monitors system memory usage using the psutil library and stores it in InfluxDB:

pip install psutil influxdb-client

import time
import psutil
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS

token = "<YOUR TOKEN HERE>"
org = "my-org"
bucket = "monitoring"

client = InfluxDBClient(url="http://localhost:8086", token=token, org=org)
write_api = client.write_api(write_options=SYNCHRONOUS)

try:
    while True:
        mem = psutil.virtual_memory()

        point = (
            Point("mem")
            .tag("host", "my-computer")
            .field("total", mem.total)
            .field("available", mem.available)
            .field("used", mem.used)
            .field("used_percent", mem.percent)
        )

        write_api.write(bucket=bucket, record=point)
        print(f"Memory: {mem.percent}% used ({mem.used / (1024**3):.1f} GB / {mem.total / (1024**3):.1f} GB)")

        time.sleep(5)
except KeyboardInterrupt:
    print("Stopped.")
finally:
    client.close()

After running this for a few minutes, go to the InfluxDB Data Explorer and create a graph showing memory usage over time. You can save the visualization to a Dashboard for continuous monitoring.

---

Further Reading

• InfluxDB Python Client Documentation
• InfluxDB 2 Write API
• InfluxDB 2 Query with Python