grafana-influxdb/influxdb-test/create-test-data.py

import json
import random
import copy

from influxdb import InfluxDBClient
import pendulum

client = InfluxDBClient(host='172.20.0.2', port=8086)

db_created = False
for db in client.get_list_database():
    if db['name'] == 'test':
        db_created = True
if not db_created:
    client.create_database('test')
client.switch_database('test')

test_data = {
    "measurement": "pocketArticles",
    "tags": {
        "user": "Nikola",
        "source": "getpocket.com"
    },
    "time": "2018-03-30T8:02:00Z",
    "fields": {
        "count": 1
    }
}

final_data = []
start = pendulum.datetime(2020, 1, 1)
end = pendulum.today()
period = pendulum.period(start, end)
for dt in period.range('hours', 2):
    num_articles = random.randrange(1, 50)
    new_point = copy.deepcopy(test_data)
    new_point['time'] = dt.isoformat()
    new_point['fields']['count'] = num_articles
    final_data.append(new_point)

if client.write_points(final_data):
    print('ok')

res = client.query('SELECT * FROM "pocketArticles"')
print(res.raw)
points = res.get_points(tags={'user': 'Nikola'})
for point in points:
    print("Time: %s, Articles: %i" % (point['time'], point['count']))