Class InfluxDBWrapper

api/app_analytics/influxdb_wrapper.py:45–154

class InfluxDBWrapper:
    client = None

    def __init__(self, name):  # type: ignore[no-untyped-def]
        self.name = name
        self.records = []

    @classmethod
    @functools.cache
    def get_client(cls) -> InfluxDBClient:
        """A singleton InfluxDB client instance"""
        retries = Retry(connect=3, read=3, redirect=3)
        return InfluxDBClient(
            url=settings.INFLUXDB_URL,
            token=settings.INFLUXDB_TOKEN,
            org=settings.INFLUXDB_ORG,
            retries=retries,
            timeout=30000,  # Hard stop to prevent hanging requests
        )

    @classmethod
    def get_downsampled_bucket(cls, size: DownsampleSize) -> str:
        return f"{settings.INFLUXDB_BUCKET}_downsampled_{size}"

    @classmethod
    def select_downsampled_bucket(cls, date_start: datetime) -> str:
        if (timezone.now() - date_start).days > 10:
            return cls.get_downsampled_bucket(DownsampleSize.ONE_HOUR)
        return cls.get_downsampled_bucket(DownsampleSize.FIFTEEN_MINUTES)

    def add_data_point(
        self,
        field_name: str,
        field_value: str | int | float,
        tags: typing.Mapping[
            str,
            str | int | float,
        ]
        | None = None,
    ) -> None:
        point = Point(self.name)
        point.field(field_name, field_value)

        if tags is not None:
            for tag_key, tag_value in tags.items():
                point = point.tag(tag_key, tag_value)

        self.records.append(point)

    def write(self) -> None:
        """Persist collected data points to InfluxDB"""
        try:
            self.get_client().write_api(write_options=SYNCHRONOUS).write(
                bucket=settings.INFLUXDB_BUCKET,
                record=self.records,
            )
        except (HTTPError, InfluxDBError) as e:
            logger.warning(