Skip to content

bulletin_rest_client.py

__all__ = ['BulletinRestClient'] module-attribute

BulletinRestClient

REST client for handling Bulletin messages.

Source code in solnlib/bulletin_rest_client.py 
 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
147
148
149
150
151
class BulletinRestClient:
    """REST client for handling Bulletin messages."""

    MESSAGES_ENDPOINT = "/services/messages"

    headers = [("Content-Type", "application/json")]

    class Severity:
        INFO = "info"
        WARNING = "warn"
        ERROR = "error"

    def __init__(
        self,
        message_name: str,
        session_key: str,
        app: str,
        **context: dict,
    ):
        """Initializes BulletinRestClient.
            When creating a new bulletin message, you must provide a name, which is a kind of ID.
            If you try to create another message with the same name (ID), the API will not add another message
            to the bulletin, but it will overwrite the existing one. Similar behaviour applies to deletion.
            To delete a message, you must indicate the name (ID) of the message.
            To provide better and easier control over bulletin messages, this client works in such a way
            that there is one instance responsible for handling one specific message.
            If you need to add another message to bulletin create another instance
            with a different 'message_name'
            e.g.
            msg_1 = BulletinRestClient("message_1", "<some session key>")
            msg_2 = BulletinRestClient("message_2", "<some session key>")

        Arguments:
            message_name: Name of the message in the Splunk's bulletin.
            session_key: Splunk access token.
            app: App name of namespace.
            context: Other configurations for Splunk rest client.
        """

        self.message_name = message_name
        self.session_key = session_key
        self.app = app

        self._rest_client = rest_client.SplunkRestClient(
            self.session_key, app=self.app, **context
        )

    def create_message(
        self,
        msg: str,
        severity: Severity = Severity.WARNING,
        capabilities: Optional[List[str]] = None,
        roles: Optional[List] = None,
    ):
        """Creates a message in the Splunk's bulletin. Calling this method
        multiple times for the same instance will overwrite existing message.

        Arguments:
            msg: The message which will be displayed in the Splunk's bulletin
            severity: Severity level of the message. It has to be one of: 'info', 'warn', 'error'.
                If wrong severity is given, ValueError will be raised.
            capabilities: One or more capabilities that users must have to view the message.
                Capability names are validated.
                This argument should be provided as a list of string/s e.g. capabilities=['one', 'two'].
                If a non-existent capability is used, HTTP 400 BAD REQUEST exception will be raised.
                If argument is not a List[str] ValueError will be raised.
            roles: One or more roles that users must have to view the message. Role names are validated.
                This argument should be provided as a list of string/s e.g. roles=['user', 'admin'].
                If a non-existent role is used, HTTP 400 BAD REQUEST exception will be raised.
                If argument is not a List[str] ValueError will be raised.
        """
        body = {
            "name": self.message_name,
            "value": msg,
            "severity": severity,
            "capability": [],
            "role": [],
        }

        if severity not in (
            self.Severity.INFO,
            self.Severity.WARNING,
            self.Severity.ERROR,
        ):
            raise ValueError(
                "Severity must be one of ("
                "'BulletinRestClient.Severity.INFO', "
                "'BulletinRestClient.Severity.WARNING', "
                "'BulletinRestClient.Severity.ERROR'"
                ")."
            )

        if capabilities:
            body["capability"] = self._validate_and_get_body_value(
                capabilities, "Capabilities must be a list of strings."
            )

        if roles:
            body["role"] = self._validate_and_get_body_value(
                roles, "Roles must be a list of strings."
            )

        self._rest_client.post(self.MESSAGES_ENDPOINT, body=body, headers=self.headers)

    def get_message(self):
        """Get specific message created by this instance."""
        endpoint = f"{self.MESSAGES_ENDPOINT}/{self.message_name}"
        response = self._rest_client.get(endpoint, output_mode="json").body.read()
        return json.loads(response)

    def get_all_messages(self):
        """Get all messages in the bulletin."""
        response = self._rest_client.get(
            self.MESSAGES_ENDPOINT, output_mode="json"
        ).body.read()
        return json.loads(response)

    def delete_message(self):
        """Delete specific message created by this instance."""
        endpoint = f"{self.MESSAGES_ENDPOINT}/{self.message_name}"
        self._rest_client.delete(endpoint)

    @staticmethod
    def _validate_and_get_body_value(arg, error_msg) -> List:
        if type(arg) is list and (all(isinstance(el, str) for el in arg)):
            return [el for el in arg]
        else:
            raise ValueError(error_msg)

MESSAGES_ENDPOINT = '/services/messages' class-attribute instance-attribute

app = app instance-attribute

headers = [('Content-Type', 'application/json')] class-attribute instance-attribute

message_name = message_name instance-attribute

session_key = session_key instance-attribute

Severity

Source code in solnlib/bulletin_rest_client.py 
31
32
33
34
class Severity:
    INFO = "info"
    WARNING = "warn"
    ERROR = "error"

ERROR = 'error' class-attribute instance-attribute

INFO = 'info' class-attribute instance-attribute

WARNING = 'warn' class-attribute instance-attribute

__init__(message_name, session_key, app, **context)

Initializes BulletinRestClient. When creating a new bulletin message, you must provide a name, which is a kind of ID. If you try to create another message with the same name (ID), the API will not add another message to the bulletin, but it will overwrite the existing one. Similar behaviour applies to deletion. To delete a message, you must indicate the name (ID) of the message. To provide better and easier control over bulletin messages, this client works in such a way that there is one instance responsible for handling one specific message. If you need to add another message to bulletin create another instance with a different ‘message_name’ e.g. msg_1 = BulletinRestClient(“message_1”, ““) msg_2 = BulletinRestClient(“message_2”, ““)

Parameters:

Name Type Description Default
message_name str

Name of the message in the Splunk’s bulletin.

 required
session_key str

Splunk access token.

 required
app str

App name of namespace.

 required
context dict

Other configurations for Splunk rest client.

 {}
Source code in solnlib/bulletin_rest_client.py 
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
def __init__(
    self,
    message_name: str,
    session_key: str,
    app: str,
    **context: dict,
):
    """Initializes BulletinRestClient.
        When creating a new bulletin message, you must provide a name, which is a kind of ID.
        If you try to create another message with the same name (ID), the API will not add another message
        to the bulletin, but it will overwrite the existing one. Similar behaviour applies to deletion.
        To delete a message, you must indicate the name (ID) of the message.
        To provide better and easier control over bulletin messages, this client works in such a way
        that there is one instance responsible for handling one specific message.
        If you need to add another message to bulletin create another instance
        with a different 'message_name'
        e.g.
        msg_1 = BulletinRestClient("message_1", "<some session key>")
        msg_2 = BulletinRestClient("message_2", "<some session key>")

    Arguments:
        message_name: Name of the message in the Splunk's bulletin.
        session_key: Splunk access token.
        app: App name of namespace.
        context: Other configurations for Splunk rest client.
    """

    self.message_name = message_name
    self.session_key = session_key
    self.app = app

    self._rest_client = rest_client.SplunkRestClient(
        self.session_key, app=self.app, **context
    )

create_message(msg, severity=Severity.WARNING, capabilities=None, roles=None)

Creates a message in the Splunk’s bulletin. Calling this method multiple times for the same instance will overwrite existing message.

Parameters:

Name Type Description Default
msg str

The message which will be displayed in the Splunk’s bulletin

 required
severity Severity

Severity level of the message. It has to be one of: ‘info’, ‘warn’, ‘error’. If wrong severity is given, ValueError will be raised.

 Severity.WARNING
capabilities Optional[List[str]]

One or more capabilities that users must have to view the message. Capability names are validated. This argument should be provided as a list of string/s e.g. capabilities=[‘one’, ‘two’]. If a non-existent capability is used, HTTP 400 BAD REQUEST exception will be raised. If argument is not a List[str] ValueError will be raised.

 None
roles Optional[List]

One or more roles that users must have to view the message. Role names are validated. This argument should be provided as a list of string/s e.g. roles=[‘user’, ‘admin’]. If a non-existent role is used, HTTP 400 BAD REQUEST exception will be raised. If argument is not a List[str] ValueError will be raised.

 None
Source code in solnlib/bulletin_rest_client.py 
 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
def create_message(
    self,
    msg: str,
    severity: Severity = Severity.WARNING,
    capabilities: Optional[List[str]] = None,
    roles: Optional[List] = None,
):
    """Creates a message in the Splunk's bulletin. Calling this method
    multiple times for the same instance will overwrite existing message.

    Arguments:
        msg: The message which will be displayed in the Splunk's bulletin
        severity: Severity level of the message. It has to be one of: 'info', 'warn', 'error'.
            If wrong severity is given, ValueError will be raised.
        capabilities: One or more capabilities that users must have to view the message.
            Capability names are validated.
            This argument should be provided as a list of string/s e.g. capabilities=['one', 'two'].
            If a non-existent capability is used, HTTP 400 BAD REQUEST exception will be raised.
            If argument is not a List[str] ValueError will be raised.
        roles: One or more roles that users must have to view the message. Role names are validated.
            This argument should be provided as a list of string/s e.g. roles=['user', 'admin'].
            If a non-existent role is used, HTTP 400 BAD REQUEST exception will be raised.
            If argument is not a List[str] ValueError will be raised.
    """
    body = {
        "name": self.message_name,
        "value": msg,
        "severity": severity,
        "capability": [],
        "role": [],
    }

    if severity not in (
        self.Severity.INFO,
        self.Severity.WARNING,
        self.Severity.ERROR,
    ):
        raise ValueError(
            "Severity must be one of ("
            "'BulletinRestClient.Severity.INFO', "
            "'BulletinRestClient.Severity.WARNING', "
            "'BulletinRestClient.Severity.ERROR'"
            ")."
        )

    if capabilities:
        body["capability"] = self._validate_and_get_body_value(
            capabilities, "Capabilities must be a list of strings."
        )

    if roles:
        body["role"] = self._validate_and_get_body_value(
            roles, "Roles must be a list of strings."
        )

    self._rest_client.post(self.MESSAGES_ENDPOINT, body=body, headers=self.headers)

delete_message()

Delete specific message created by this instance.

Source code in solnlib/bulletin_rest_client.py 
141
142
143
144
def delete_message(self):
    """Delete specific message created by this instance."""
    endpoint = f"{self.MESSAGES_ENDPOINT}/{self.message_name}"
    self._rest_client.delete(endpoint)

get_all_messages()

Get all messages in the bulletin.

Source code in solnlib/bulletin_rest_client.py 
134
135
136
137
138
139
def get_all_messages(self):
    """Get all messages in the bulletin."""
    response = self._rest_client.get(
        self.MESSAGES_ENDPOINT, output_mode="json"
    ).body.read()
    return json.loads(response)

get_message()

Get specific message created by this instance.

Source code in solnlib/bulletin_rest_client.py 
128
129
130
131
132
def get_message(self):
    """Get specific message created by this instance."""
    endpoint = f"{self.MESSAGES_ENDPOINT}/{self.message_name}"
    response = self._rest_client.get(endpoint, output_mode="json").body.read()
    return json.loads(response)