Learn more  » Push, build, and install  RubyGems npm packages Python packages Maven artifacts PHP packages Go Modules Bower components Debian packages RPM packages NuGet packages

agriconnect / influxalchemy   python

Repository URL to install this package:

/ meta.py

""" InfluxDB Meta Measurement. """
from datetime import date

from . import operations

try:
    from datetime import timezone
    UTC = timezone.utc  # pragma: no cover
except ImportError:     # pragma: no cover
    import pytz         # pragma: no cover
    UTC = pytz.utc      # pragma: no cover


def make_tz_aware(datetime_obj):
    """ Make a date/datetime object to be timezone-aware """
    # 'date' object doesn't need to be timezone aware
    # pylint: disable=unidiomatic-typecheck
    if type(datetime_obj) is date:
        return datetime_obj
    # Already aware
    if datetime_obj.tzinfo:
        return datetime_obj
    # With naive datetime object, assume it is UTC
    return datetime_obj.replace(tzinfo=UTC)


class MetaMeasurement(type):
    """ Meta class of Measurement. """
    def __new__(cls, name, bases, dict_):
        dict_.setdefault("__measurement__", name)
        return super(MetaMeasurement, cls).__new__(cls, name, bases, dict_)

    def __getattribute__(cls, name):
        try:
            return super(MetaMeasurement, cls).__getattribute__(name)
        except AttributeError:
            if name == "time":
                return Time(name, cls)
            return Tag(name, cls)

    def __str__(cls):
        return cls.__measurement__

    def __eq__(cls, other):
        return str(cls) == str(other)

    def __ne__(cls, other):
        return str(cls) != str(other)

    def __or__(cls, other):
        left = str(cls).strip("/")
        name = "_".join(left.split("|") + [str(other)])
        bases = (cls,)
        measurement = "/%s|%s/" % (str(cls).strip("/"), other)
        return type(name, bases, {"__measurement__": measurement})

    def __hash__(cls):
        return id(cls)

    @property
    def measurement(cls):
        """ Reflexive reference to measurement. """
        return cls


class Tag(object):
    """ InfluxDB Tag instance.

        name        (str):          Name of Tag
        measurement (Measurement):  Measurement of tag
    """
    def __init__(self, name, measurement):
        self._name = name
        self.measurement = measurement

    def __str__(self):
        return self._name

    def __repr__(self):
        return "<%s.%s>" % (self.measurement.__measurement__, self._name)

    def __eq__(self, other):
        return TagExp.equals(self, other)

    def __ne__(self, other):
        return TagExp.notequals(self, other)

    def __gt__(self, other):
        return TagExp.greater_than(self, other)

    def __lt__(self, other):
        return TagExp.less_than(self, other)

    def __ge__(self, other):
        return TagExp.greater_equal(self, other)

    def __le__(self, other):
        return TagExp.less_equal(self, other)

    def like(self, other):
        """ self =~ other """
        return TagExp.like(self, other)

    def notlike(self, other):
        """ self !~ other """
        return TagExp.notlike(self, other)


class Time(Tag):
    """ Time of InfluxDB Measurement. """
    def between(self, start, end, startinc=True, endinc=True):
        """ Query times between extremes.

            Arguments:
                start    (str):      Start of time
                end      (str):      End of time
                startinc (boolean):  Start-inclusive flag
                endinc   (boolean):  End-inclusive flag

            Returns:
                Time expression.
        """
        if startinc is True:
            startexp = TagExp.greater_equal(self, start)
        else:
            startexp = TagExp.greater_than(self, start)
        if endinc is True:
            endexp = TagExp.less_equal(self, end)
        else:
            endexp = TagExp.less_than(self, end)
        return startexp & endexp


class TagExp(object):
    """ A tag query expression. """
    def __init__(self, left, op, right):
        self._left = left
        self._op = op
        lits = [operations.LK, operations.NK, operations.AND, operations.OR]
        if self._op in lits or isinstance(left, Time):
            # If the right-hand-side value is a date/datetime object,
            # we convert it to RFC3339 representation
            # and wrap inside single quote
            if isinstance(right, date):
                right = repr(make_tz_aware(right).isoformat())
            self._right = right
        else:
            self._right = repr(right)

    def __str__(self):
        return "%s%s%s" % (self._left, self._op, self._right)

    def __repr__(self):
        return "[ %s ]" % self

    def __eq__(self, other):
        return str(self) == str(other)

    def __ne__(self, other):
        return str(self) != str(other)

    def __and__(self, other):
        return TagExp(str(self), " AND ", str(other))

    def __or__(self, other):
        return TagExp(str(self), " OR ", str(other))

    def __invert__(self):
        return TagExp(self._left, ~self._op, self._right)

    @classmethod
    def equals(cls, self, other):
        """ left = right """
        return cls(self, operations.EQ, other)

    @classmethod
    def notequals(cls, self, other):
        """ left != right """
        return cls(self, operations.NE, other)

    @classmethod
    def greater_than(cls, self, other):
        """ left > right """
        return cls(self, operations.GT, other)

    @classmethod
    def less_than(cls, self, other):
        """ left < right """
        return cls(self, operations.LT, other)

    @classmethod
    def greater_equal(cls, self, other):
        """ left >= right """
        return cls(self, operations.GE, other)

    @classmethod
    def less_equal(cls, self, other):
        """ left <= right """
        return cls(self, operations.LE, other)

    @classmethod
    def like(cls, self, other):
        """ left =~ right """
        return cls(self, operations.LK, other)

    @classmethod
    def notlike(cls, self, other):
        """ left !~ right """
        return cls(self, operations.NK, other)