Method get_client

api/app_analytics/influxdb_wrapper.py:54–63

A singleton InfluxDB client instance

(cls)

    @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: