Use official influxdb client

linien-common/linien_common/influxdb.py

@@ -17,6 +17,7 @@
 
 import json
 import logging
+from dataclasses import dataclass
 
 from .config import USER_DATA_PATH
 
@@ -26,26 +27,13 @@
 logger.setLevel(logging.DEBUG)
 
 
+@dataclass
 class InfluxDBCredentials:
-    def __init__(
-        self,
-        url: str = "http://localhost:8086",
-        org: str = "my-org",
-        token: str = "my-token",
-        bucket: str = "my-bucket",
-        measurement: str = "my-measurement",
-    ) -> None:
-        self.url = url
-        self.org = org
-        self.token = token
-        self.bucket = bucket
-        self.measurement = measurement
-
-    def __str__(self) -> str:
-        return (
-            f"url: {self.url}, org: {self.org}, token: {self.token}, "
-            f" bucket: {self.bucket}, measurement: {self.measurement}"
-        )
+    url: str = "http://localhost:8086"
+    org: str = "my-org"
+    token: str = "my-token"
+    bucket: str = "my-bucket"
+    measurement: str = "my-measurement"
 
 
 def save_credentials(credentials: InfluxDBCredentials) -> None:

linien-server/linien_server/influxdb.py

@@ -18,7 +18,9 @@
 from threading import Event, Thread
 from time import sleep
 
-import requests
+from influxdb_client import InfluxDBClient
+from influxdb_client.client.write_api import SYNCHRONOUS
+from linien_common.communication import ParameterValues
 from linien_common.influxdb import InfluxDBCredentials, save_credentials
 from linien_server.parameters import Parameters
 
@@ -27,10 +29,11 @@ class InfluxDBLogger:
     def __init__(
         self, credentials: InfluxDBCredentials, parameters: Parameters
     ) -> None:
-        self.credentials = credentials
-        self.parameters = parameters
+        self.credentials: InfluxDBClient = credentials
+        self.parameters: Parameters = parameters
         self.stop_event = Event()
         self.stop_event.set()
+        self.update_connection()
 
     @property
     def credentials(self) -> InfluxDBCredentials:
@@ -41,6 +44,14 @@ def credentials(self, value: InfluxDBCredentials) -> None:
         self._credentials = value
         save_credentials(value)
 
+    def update_connection(self) -> InfluxDBClient:
+        client = InfluxDBClient(
+            url=self.credentials.url,
+            token=self.credentials.token,
+            org=self.credentials.org,
+        )
+        self.write_api = client.write_api(write_options=SYNCHRONOUS)
+
     def start_logging(self, interval: float) -> None:
         conn_success, status_code, message = self.test_connection(self.credentials)
         self.thread = Thread(
@@ -78,46 +89,27 @@ def test_connection(
         self, credentials: InfluxDBCredentials
     ) -> tuple[bool, int, str]:
         """Write empty data to the server to test the connection"""
-        try:
-            response = self.write_data(credentials, data={})
-            success = response.status_code == 204
-            status_code = response.status_code
-            text = response.text
-        except requests.exceptions.ConnectionError:
-            success = False
-            status_code = 404
-            text = "Failed to establish connection."
-        return success, status_code, text
+        client = InfluxDBClient(
+            url=credentials.url,
+            token=credentials.token,
+            org=credentials.org,
+        )
+
+        # FIXME: This does not test the credentials, yet.
+        status_code = 0
+        message = ""
+        success = client.ping()
+        return success, status_code, message
 
     def write_data(
-        self, credentials: InfluxDBCredentials, data: dict
-    ) -> requests.Response:
+        self, credentials: InfluxDBCredentials, fields: dict[str, ParameterValues]
+    ) -> None:
         """Write data to the database"""
-        endpoint = credentials.url + "/api/v2/write"
-        headers = {
-            "Authorization": "Token " + credentials.token,
-            "Content-Type": "text/plain; charset=utf-8",
-            "Accept": "application/json",
-        }
-        params = {
-            "org": credentials.org,
-            "bucket": credentials.bucket,
-            "precision": "ns",
-        }
-
-        point = self._convert_to_line_protocol(data)
-
-        response = requests.post(endpoint, headers=headers, params=params, data=point)
-        return response
-
-    def _convert_to_line_protocol(self, data: dict) -> str:
-        if not data:
-            return ""
-        point = self.credentials.measurement
-        for i, (key, value) in enumerate(data.items()):
-            if i == 0:
-                point += " "
-            else:
-                point += ","
-            point += f"{key}={value}"
-        return point
+        self.write_api.write(
+            bucket=credentials.bucket,
+            org=credentials.org,
+            record={
+                "measurement": credentials.measurement,
+                "fields": fields,
+            },
+        )

linien-server/pyproject.toml

@@ -24,9 +24,9 @@ classifiers = [
 requires-python = ">=3.10"
 dependencies = [
     "cma>=3.0.3",
+    "influxdb-client[ciso]>=1.9,<2.0",
     "pylpsd>=0.1.4",
     "pyrp3>=2.0.0,<3.0;platform_machine=='armv7l'",
-    "requests>=2.25.1",
     "linien-common==2.0.0.dev0",
 ]
 

pyproject.toml

@@ -43,5 +43,6 @@ module = [
   "pylpsd.*",
   "pyqtgraph.*",
   "fire",
+  "influxdb_client.*",
 ]
 ignore_missing_imports = true