# Luna Streams v.1.1.15 documentation

## luna_streams.crutches_on_wheels.cow.monitoring.influx_adapter

# Source code for luna_streams.crutches_on_wheels.cow.monitoring.influx_adapter

    """
    Module contains classes for sending a data to an influx monitoring.
    """

    # Ignore no_pandas_warning in aioinflux
    import warnings
    from dataclasses import dataclass
    from typing import Iterable

    import aiohttp

    warnings.simplefilter("ignore", UserWarning)
    from aiohttp.web_exceptions import HTTPException  # pylint: disable-msg=C0413

    warnings.simplefilter("default", UserWarning)

    from influxdb_client.rest import ApiException  # pylint: disable-msg=C0413

    from .base_adapter import BaseMonitoringAdapter  # pylint: disable-msg=C0413
    from .helpers import lineProtocolFieldsGenerator  # pylint: disable-msg=C0413
    from .points import BaseMonitoringPoint  # pylint: disable-msg=C0413

    @dataclass
    class InfluxSettings:
        """Container for influx 2.x settings"""

        # influx base url
        url: str
        # bucket name
        bucket: str
        # workspace
        organization: str
        # authentication token
        token: str

    class InfluxMonitoringAdapter(BaseMonitoringAdapter):
        """
        Influx 2.x adaptor. Suspended to send points to an influxdb

        Attributes:
            bucket (str): influx bucket name

        """

        _buffer: list[str]

        def __init__(self, settings: InfluxSettings, flushingPeriod: int):
            super().__init__(flushingPeriod=flushingPeriod)
            self._influxSettings = settings
            self._buffer = []
            self.bucket = settings.bucket
            self.session = aiohttp.ClientSession()

        def addPointsToBuffer(self, points: Iterable[BaseMonitoringPoint]) -> None:
            """
            Add points to buffer.

            Args:
                points: points
            """
            self._buffer.extend((self.generatePointStr(point) for point in points))

        def generatePointStr(self, point: BaseMonitoringPoint) -> str:
            """
            Generate string from point

            Args:
                point: point

            Returns:
                influx line protocol string
            """
            time = int(point.eventTime * 10**9)
            measurement = point.series
            tags = ",".join([f"{tag}={value}" for tag, value in point.tags.items()])
            if hasattr(point, "convertFieldsToInfluxLineProtocol"):
                fields = point.convertFieldsToInfluxLineProtocol()
            else:
                fields = self.convertFieldsToInfluxLineProtocol(point.fields)
            pointStr = f"{measurement},{tags} {fields} {time}"
            return pointStr

        @staticmethod
        def convertFieldsToInfluxLineProtocol(fields: dict) -> str:
            """
            Convert field value to influx line protocol format

            Args:
                fields: dict with values to convert

            Returns:
                line protocol string
            """
            return ",".join(lineProtocolFieldsGenerator(fields))

        async def _writePoints(self, points):
            """
            Write points.
            Args:
                points
            """
            try:
                headers = {
                    "Content-Type": "text/plain",
                    "Authorization": f"Token {self._influxSettings.token}",
                }
                params = {"bucket": self.bucket, "org": self._influxSettings.organization}

                async def fetch(session, data):
                    resp = await session.post(
                        f"{self._influxSettings.url}/api/v2/write", data=data, headers=headers, params=params
                    )
                    return resp

                data = "\n".join(points)
                return await fetch(self.session, data.encode())
            except (ApiException, HTTPException) as exc:
                self.logger.warning(f"failed write monitoring buffering points, message: {exc}")

        def clearBuffer(self) -> list[str]:
            """Clear monitoring buffer"""
            copyPoints = self._buffer.copy()
            self._buffer = []
            return copyPoints

        async def stopMonitoring(self) -> None:
            """
            Stop monitoring (cancel all request and stop getting new).
            """
            await super().stopMonitoring()
            await self.session.close()