File: _health.py

package info (click to toggle)
python-gvm 26.5.0-1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 5,132 kB
  • sloc: python: 44,662; makefile: 18
file content (87 lines) | stat: -rw-r--r-- 2,607 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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
# SPDX-FileCopyrightText: 2025 Greenbone AG
#
# SPDX-License-Identifier: GPL-3.0-or-later

"""
API wrapper for accessing the /health endpoints of the openvasd HTTP API.
"""

import httpx

from ._api import OpenvasdAPI


class HealthAPI(OpenvasdAPI):
    """
    Provides access to the openvasd /health endpoints, which expose the
    operational state of the scanner.

    All methods return the HTTP status code of the response and raise an exception
    if the server returns an error response (4xx or 5xx).
    """

    def get_alive(self) -> int:
        """
        Check if the scanner process is alive.

        Returns:
            HTTP status code (e.g., 200 if alive).

        Raises:
            httpx.HTTPStatusError: If the server response indicates failure and
                exceptions are not suppressed.

        See: GET /health/alive in the openvasd API documentation.
        """
        try:
            response = self._client.get("/health/alive")
            response.raise_for_status()
            return response.status_code
        except httpx.HTTPStatusError as e:
            if self._suppress_exceptions:
                return e.response.status_code
            raise

    def get_ready(self) -> int:
        """
        Check if the scanner is ready to accept requests (e.g., feed loaded).

        Returns:
            HTTP status code (e.g., 200 if ready).

        Raises:
            httpx.HTTPStatusError: If the server response indicates failure and
                exceptions are not suppressed.

        See: GET /health/ready in the openvasd API documentation.
        """
        try:
            response = self._client.get("/health/ready")
            response.raise_for_status()
            return response.status_code
        except httpx.HTTPStatusError as e:
            if self._suppress_exceptions:
                return e.response.status_code
            raise

    def get_started(self) -> int:
        """
        Check if the scanner has fully started.

        Returns:
            HTTP status code (e.g., 200 if started).

        Raises:
            httpx.HTTPStatusError: If the server response indicates failure and
                exceptions are not suppressed.

        See: GET /health/started in the openvasd API documentation.
        """
        try:
            response = self._client.get("/health/started")
            response.raise_for_status()
            return response.status_code
        except httpx.HTTPStatusError as e:
            if self._suppress_exceptions:
                return e.response.status_code
            raise