diff --git a/gso/services/librenms_client.py b/gso/services/librenms_client.py index c700c92fc349afdb6521c65d0ef6720548f78597..a4857b9d4a59eb5304fdea193c756ad42a2e09c8 100644 --- a/gso/services/librenms_client.py +++ b/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()