FM / mtrlib

/opt/imh-python/lib/python3.9/site-packages/mtrlib/

/opt/imh-python/lib/python3.9/site-packages/mtrlib

Name	Size	Mode	Actions
carbon.py	2003	0644	edit dl rm
graphite.py	6515	0644	edit dl rm
nagios.py	1299	0644	edit dl rm
nscaweb.py	3960	0644	edit dl rm
__init__.py	69	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/mtrlib/graphite.py (6515B)

"""Functions for Retrieving data from graphite"""
from typing import Union, List, Optional, Tuple
from datetime import datetime
import urllib.parse
import requests
from arrow.arrow import Arrow
import arrow

QUERY_FMT = '%H:%M_%Y%m%d'

class GraphiteError(Exception):
    """Raised on error when fetching graphite data"""

class Graphite:
    """Performs Graphite Queries"""

def __init__(
        self,
        render_url: Optional[str] = 'https://graphite.imhadmin.net/render/',
        timeout: Optional[Union[float, Tuple[float]]] = 600,
    ):
        self.render_url = render_url
        self.timeout = timeout

@staticmethod
    def _params(
        target: Union[str, List[str]],
        start: Optional[Union[datetime, Arrow, str, int]] = None,
        stop: Optional[Union[datetime, Arrow, str, int]] = None,
        tzname: Optional[str] = None,
    ) -> dict:
        # int -> arrow
        if isinstance(start, int):
            start = arrow.get(start)
        if isinstance(stop, int):
            stop = arrow.get(stop)
        # format datetimes
        if isinstance(start, (datetime, Arrow)):
            if tzname is None:  # if tzname was not overridden
                tzname = start.tzname()
            start = start.strftime(QUERY_FMT)
        if isinstance(stop, (datetime, Arrow)):
            stop = stop.strftime(QUERY_FMT)
        # graphite's pytz may be old
        if tzname in ('PDT', 'PST'):
            tzname = 'America/Los_Angeles'
        if tzname in ('EST', 'EDT'):
            tzname = 'America/New_York'
        params = {'target': target}
        if start is not None:
            params['from'] = start
        if stop is not None:
            params['until'] = stop
        if tzname is not None:
            params['tz'] = tzname
        return params

def query(
        self,
        target: Union[str, List[str]],
        start: Optional[Union[datetime, Arrow, str, int]] = None,
        stop: Optional[Union[datetime, Arrow, str, int]] = None,
        tzname: Optional[str] = None,
    ) -> list:
        """Query data from graphite

Args:
            target (str or list): graphite bucket(s) to fetch data for
            start: start time to fetch data for. If not supplied, it'll
                be 24 hours ago **(see note)**
            stop: stop time to fetch data for. If not supplied, it'll be now
                **(see note)**
            tzname: timezone to use. If not supplied, the ``start`` arg's
                timezone will be used. If the ``start`` arg is not timezone
                aware, it'll use graphite defaults

Raises:
            GraphiteError: on any error

Note:
            ``start`` and ``stop`` will accept ``arrow`` objects, ``datetimes``,
            ``int`` for unix timestamps, and ``str``. If it's a ``str``, it'll
            be sent as a literal and handled per graphite's documentation
            https://graphite.readthedocs.io/en/latest/render_api.html
        """
        params = self._params(target, start, stop, tzname)
        params['format'] = 'json'
        try:
            req = requests.get(
                self.render_url,
                timeout=self.timeout,
                params=params,
            )
        except requests.RequestException as exc:
            raise GraphiteError(exc) from exc
        try:
            return req.json()
        except ValueError as exc:
            if req.status_code != 200:
                raise GraphiteError(
                    f"{req.url}\nHTTP {req.status_code}: {req.reason}"
                ) from exc
            raise GraphiteError(
                f"{req.url}\n{type(exc).__name__}: {exc}"
            ) from exc

def query_dict(
        self,
        target: Union[str, List[str]],
        start: Optional[Union[datetime, Arrow, str, int]] = None,
        stop: Optional[Union[datetime, Arrow, str, int]] = None,
        tzname: Optional[str] = None,
    ) -> dict:
        """Just like ``query`` except it converts the data to a dict

Raises:
            GraphiteError: on any error

def img_url(
        self,
        target: Union[str, List[str]],
        start: Optional[Union[datetime, Arrow, str, int]] = None,
        stop: Optional[Union[datetime, Arrow, str, int]] = None,
        tzname: Optional[str] = None,
        **kwargs,
    ) -> str:
        """Gets a graphite render URL for an image but does not fetch it

Raises:
            GraphiteError: on any error

cmd: