Method write

api/app_analytics/influxdb_wrapper.py:94–111

Persist collected data points to InfluxDB

(self)

        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(
                "Failed to write records to Influx: %s",
                str(e),
                exc_info=e,
            )
            logger.debug(
                "Records: %s. Bucket: %s",
                self.records,
                settings.INFLUXDB_BUCKET,
            )

    @classmethod
    def influx_query_manager(