Source code for influxdb_client.client.flux_table

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."""


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,
            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: ( 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 = 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')}>"