Skip to content

Librenms client

The LibreNMS module interacts with the inventory management system of GAP.

LibreNMSClient

The client for LibreNMS that interacts with the inventory management system.

Source code in gso/services/librenms_client.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class LibreNMSClient:
    """The client for LibreNMS that interacts with the inventory management system."""

    def __init__(self) -> None:
        """Initialise a new LibreNMS client with an authentication token."""
        config = load_oss_params().MONITORING
        token = config.LIBRENMS.token

        self.base_url = config.LIBRENMS.base_url
        self.snmp_config = config.SNMP

        self.session = requests.Session()
        self.session.mount("https://", HTTPAdapter(max_retries=5))
        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, timeout=(10, 75))
        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.

        Args:
            fqdn: The FQDN of a device that is retrieved.

        Returns:
            A JSON formatted list of devices that match the queried FQDN.

        Raises:
            HTTPError: Raises an HTTP error 404 when the device is not found
        """
        response = self._send_request("GET", f"/devices/{fqdn}")
        response.raise_for_status()

        return response.json()

    def device_exists(self, fqdn: str) -> bool:
        """Check whether a device exists in LibreNMS.

        Args:
            fqdn: The hostname that should be checked for.

        Returns:
            Whether the device exists or not.
        """
        try:
            device = self.get_device(fqdn)
        except HTTPError as e:
            if e.response.status_code == HTTPStatus.NOT_FOUND:
                return False
            raise

        return device["status"] == "ok"

    def add_device(self, fqdn: str, snmp_version: SNMPVersion) -> dict[str, Any]:
        """Add a new device to LibreNMS.

        Args:
            fqdn: The hostname of the newly added device.
            snmp_version: The SNMP version of the new device, which decides the authentication parameters that LibreNMS
                should use to poll the device.
        """
        device_data = {
            "display": fqdn,
            "hostname": fqdn,
            "sysName": fqdn,
            "snmpver": snmp_version.value,
        }
        device_data.update(getattr(self.snmp_config, snmp_version))

        device = self._send_request("POST", "/devices", device_data)
        device.raise_for_status()

        return device.json()

    def remove_device(self, fqdn: str) -> dict[str, Any]:
        """Remove a device from LibreNMS.

        Args:
            fqdn: The FQDN of the hostname that should get deleted.

        Returns:
            A JSON representation of the device that got removed.

        Raises:
            HTTPError: Raises an exception if the request did not succeed.
        """
        device = self._send_request("DELETE", f"/devices/{fqdn}")
        device.raise_for_status()

        return device.json()

    def validate_device(self, fqdn: str) -> str | None:
        """Validate a device in LibreNMS by fetching the record match the queried FQDN against its hostname.

        Args:
            fqdn: The FQDN of the host that is validated.

        Returns:
            A list of errors, if empty the device is successfully validated.
        """
        error = None
        try:
            device = self.get_device(fqdn)
            received_hostname = device["devices"][0]["hostname"]

            if received_hostname != fqdn:
                error = (
                    f"Device hostname in LibreNMS does not match FQDN.\n"
                    f"Expected '{fqdn}' but got '{received_hostname}'."
                )
        except HTTPError as e:
            if e.response.status_code == HTTPStatus.NOT_FOUND:
                error = "Device does not exist in LibreNMS."
            else:
                raise

        return error

__init__()

Initialise a new LibreNMS client with an authentication token.

Source code in gso/services/librenms_client.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def __init__(self) -> None:
    """Initialise a new LibreNMS client with an authentication token."""
    config = load_oss_params().MONITORING
    token = config.LIBRENMS.token

    self.base_url = config.LIBRENMS.base_url
    self.snmp_config = config.SNMP

    self.session = requests.Session()
    self.session.mount("https://", HTTPAdapter(max_retries=5))
    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,
    })

get_device(fqdn)

Get an existing device from LibreNMS.

Parameters:

Name Type Description Default
fqdn str

The FQDN of a device that is retrieved.

 required

Returns:

Type Description
dict[str, Any]

A JSON formatted list of devices that match the queried FQDN.

Raises:

Type Description
HTTPError

Raises an HTTP error 404 when the device is not found
Source code in gso/services/librenms_client.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def get_device(self, fqdn: str) -> dict[str, Any]:
    """Get an existing device from LibreNMS.

    Args:
        fqdn: The FQDN of a device that is retrieved.

    Returns:
        A JSON formatted list of devices that match the queried FQDN.

    Raises:
        HTTPError: Raises an HTTP error 404 when the device is not found
    """
    response = self._send_request("GET", f"/devices/{fqdn}")
    response.raise_for_status()

    return response.json()

device_exists(fqdn)

Check whether a device exists in LibreNMS.

Parameters:

Name Type Description Default
fqdn str

The hostname that should be checked for.

 required

Returns:

Type Description
bool

Whether the device exists or not.
Source code in gso/services/librenms_client.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def device_exists(self, fqdn: str) -> bool:
    """Check whether a device exists in LibreNMS.

    Args:
        fqdn: The hostname that should be checked for.

    Returns:
        Whether the device exists or not.
    """
    try:
        device = self.get_device(fqdn)
    except HTTPError as e:
        if e.response.status_code == HTTPStatus.NOT_FOUND:
            return False
        raise

    return device["status"] == "ok"

add_device(fqdn, snmp_version)

Add a new device to LibreNMS.

Parameters:

Name Type Description Default
fqdn str

The hostname of the newly added device.

 required
snmp_version SNMPVersion

The SNMP version of the new device, which decides the authentication parameters that LibreNMS should use to poll the device.

 required
Source code in gso/services/librenms_client.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def add_device(self, fqdn: str, snmp_version: SNMPVersion) -> dict[str, Any]:
    """Add a new device to LibreNMS.

    Args:
        fqdn: The hostname of the newly added device.
        snmp_version: The SNMP version of the new device, which decides the authentication parameters that LibreNMS
            should use to poll the device.
    """
    device_data = {
        "display": fqdn,
        "hostname": fqdn,
        "sysName": fqdn,
        "snmpver": snmp_version.value,
    }
    device_data.update(getattr(self.snmp_config, snmp_version))

    device = self._send_request("POST", "/devices", device_data)
    device.raise_for_status()

    return device.json()

remove_device(fqdn)

Remove a device from LibreNMS.

Parameters:

Name Type Description Default
fqdn str

The FQDN of the hostname that should get deleted.

 required

Returns:

Type Description
dict[str, Any]

A JSON representation of the device that got removed.

Raises:

Type Description
HTTPError

Raises an exception if the request did not succeed.
Source code in gso/services/librenms_client.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def remove_device(self, fqdn: str) -> dict[str, Any]:
    """Remove a device from LibreNMS.

    Args:
        fqdn: The FQDN of the hostname that should get deleted.

    Returns:
        A JSON representation of the device that got removed.

    Raises:
        HTTPError: Raises an exception if the request did not succeed.
    """
    device = self._send_request("DELETE", f"/devices/{fqdn}")
    device.raise_for_status()

    return device.json()

validate_device(fqdn)

Validate a device in LibreNMS by fetching the record match the queried FQDN against its hostname.

Parameters:

Name Type Description Default
fqdn str

The FQDN of the host that is validated.

 required

Returns:

Type Description
str | None

A list of errors, if empty the device is successfully validated.
Source code in gso/services/librenms_client.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def validate_device(self, fqdn: str) -> str | None:
    """Validate a device in LibreNMS by fetching the record match the queried FQDN against its hostname.

    Args:
        fqdn: The FQDN of the host that is validated.

    Returns:
        A list of errors, if empty the device is successfully validated.
    """
    error = None
    try:
        device = self.get_device(fqdn)
        received_hostname = device["devices"][0]["hostname"]

        if received_hostname != fqdn:
            error = (
                f"Device hostname in LibreNMS does not match FQDN.\n"
                f"Expected '{fqdn}' but got '{received_hostname}'."
            )
    except HTTPError as e:
        if e.response.status_code == HTTPStatus.NOT_FOUND:
            error = "Device does not exist in LibreNMS."
        else:
            raise

    return error