"""
Flux employs a basic data model built from basic data types.
The data model consists of tables, records, columns.
"""
from json import JSONEncoder
class FluxStructure:
"""The data model consists of tables, records, columns."""
pass
class FluxStructureEncoder(JSONEncoder):
"""The FluxStructure encoder to encode query results to JSON."""
def default(self, obj):
"""Return serializable objects for JSONEncoder."""
import datetime
if isinstance(obj, FluxStructure):
return obj.__dict__
elif isinstance(obj, (datetime.datetime, datetime.date)):
return obj.isoformat()
return super().default(obj)
[docs]class FluxTable(FluxStructure):
"""
A table is set of records with a common set of columns and a group key.
The table can be serialized into JSON by::
import json
from influxdb_client.client.flux_table import FluxStructureEncoder
output = json.dumps(tables, cls=FluxStructureEncoder, indent=2)
print(output)
"""
def __init__(self) -> None:
"""Initialize defaults."""
self.columns = []
self.records = []
[docs] def get_group_key(self):
"""
Group key is a list of columns.
A table’s group key denotes which subset of the entire dataset is assigned to the table.
"""
return list(filter(lambda column: (column.group is True), self.columns))
def __str__(self):
"""Return formatted output."""
cls_name = type(self).__name__
return cls_name + "() columns: " + str(len(self.columns)) + ", records: " + str(len(self.records))
def __repr__(self):
"""Format for inspection."""
return f"<{type(self).__name__}: {len(self.columns)} columns, {len(self.records)} records>"
def __iter__(self):
"""Iterate over records."""
return iter(self.records)
class FluxColumn(FluxStructure):
"""A column has a label and a data type."""
def __init__(self, index=None, label=None, data_type=None, group=None, default_value=None) -> None:
"""Initialize defaults."""
self.default_value = default_value
self.group = group
self.data_type = data_type
self.label = label
self.index = index
def __repr__(self):
"""Format for inspection."""
fields = [repr(self.index)] + [
f'{name}={getattr(self, name)!r}' for name in (
'label', 'data_type', 'group', 'default_value'
) if getattr(self, name) is not None
]
return f"{type(self).__name__}({', '.join(fields)})"
[docs]class FluxRecord(FluxStructure):
"""A record is a tuple of named values and is represented using an object type."""
def __init__(self, table, values=None) -> None:
"""Initialize defaults."""
if values is None:
values = {}
self.table = table
self.values = values
[docs] def get_start(self):
"""Get '_start' value."""
return self["_start"]
[docs] def get_stop(self):
"""Get '_stop' value."""
return self["_stop"]
[docs] def get_time(self):
"""Get timestamp."""
return self["_time"]
[docs] def get_value(self):
"""Get field value."""
return self["_value"]
[docs] def get_field(self):
"""Get field name."""
return self["_field"]
[docs] def get_measurement(self):
"""Get measurement name."""
return self["_measurement"]
def __getitem__(self, key):
"""Get value by key."""
return self.values.__getitem__(key)
def __setitem__(self, key, value):
"""Set value with key and value."""
return self.values.__setitem__(key, value)
def __str__(self):
"""Return formatted output."""
cls_name = type(self).__name__
return cls_name + "() table: " + str(self.table) + ", " + str(self.values)
def __repr__(self):
"""Format for inspection."""
return f"<{type(self).__name__}: field={self.values.get('_field')}, value={self.values.get('_value')}>"