Re-use session in LibreNMS client

gso/services/librenms_client.py

@@ -3,10 +3,10 @@
 import logging
 from http import HTTPStatus
 from importlib import metadata
-from typing import Any
+from typing import Any, Literal
 
 import requests
-from requests import HTTPError
+from requests import HTTPError, Response
 
 from gso.settings import load_oss_params
 from gso.utils.helpers import SNMPVersion
@@ -25,12 +25,23 @@ class LibreNMSClient:
         self.base_url = config.LIBRENMS.base_url
         self.snmp_config = config.SNMP
 
-        self.headers = {
+        self.session = requests.Session()
+        self.session.headers.update({
             "User-Agent": f"geant-service-orchestrator/{metadata.version("geant-service-orchestrator")}",
             "Accept": "application/json",
             "Content-Type": "application/json",
             "X-Auth-Token": token,
-        }
+        })
+
+    def _send_request(
+            self, method: Literal["GET", "POST", "PUT", "DELETE"], endpoint: str, data: dict[str, Any] | None = None
+    ) -> Response:
+        url = self.base_url + endpoint
+        logger.debug("LibreNMS - Sending request", extra={"method": method, "endpoint": url, "form_data": data})
+        result = self.session.request(method, url, json=data)
+        logger.debug("LibreNMS - Received response", extra=result.__dict__)
+
+        return result
 
     def get_device(self, fqdn: str) -> dict[str, Any]:
         """Get an existing device from LibreNMS.
@@ -39,7 +50,7 @@ class LibreNMSClient:
         :return dict[str, Any]: A :term:`JSON` formatted list of devices that match the queried :term:`FQDN`.
         :raises HTTPError: Raises an HTTP error 404 when the device is not found
         """
-        response = requests.get(f"{self.base_url}/devices/{fqdn}", headers=self.headers, timeout=(0.5, 75))
+        response = self._send_request("GET", f"/devices/{fqdn}")
         response.raise_for_status()
 
         return response.json()
@@ -74,7 +85,7 @@ class LibreNMSClient:
         }
         device_data.update(getattr(self.snmp_config, snmp_version))
 
-        device = requests.post(f"{self.base_url}/devices", headers=self.headers, json=device_data, timeout=(0.5, 75))
+        device = self._send_request("POST", "/devices", device_data)
         device.raise_for_status()
 
         return device.json()
@@ -86,7 +97,7 @@ class LibreNMSClient:
         :return dict[str, Any]: A JSON representation of the device that got removed.
         :raises HTTPError: Raises an exception if the request did not succeed.
         """
-        device = requests.delete(f"{self.base_url}/devices/{fqdn}", headers=self.headers, timeout=(0.5, 75))
+        device = self._send_request("DELETE", f"/devices/{fqdn}")
         device.raise_for_status()
 
         return device.json()