server-client-python/tableauserverclient/server/endpoint/data_alert_endpoint.py at ce4d1f54b50eb1d8674ce516bc4fe1340b5cd8d7 · tableau/server-client-python · GitHub

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
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
import logging

from .endpoint import api, Endpoint
from .exceptions import MissingRequiredFieldError
from tableauserverclient.server import RequestFactory
from tableauserverclient.models import DataAlertItem, PaginationItem, UserItem

from tableauserverclient.helpers.logging import logger

from typing import Optional, TYPE_CHECKING, Union


if TYPE_CHECKING:
    from ..server import Server
    from ..request_options import RequestOptions


class DataAlerts(Endpoint):
    def __init__(self, parent_srv: "Server") -> None:
        super().__init__(parent_srv)

    @property
    def baseurl(self) -> str:
        return f"{self.parent_srv.baseurl}/sites/{self.parent_srv.site_id}/dataAlerts"

    @api(version="3.2")
    def get(self, req_options: Optional["RequestOptions"] = None) -> tuple[list[DataAlertItem], PaginationItem]:
        logger.info("Querying all dataAlerts on site")
        url = self.baseurl
        server_response = self.get_request(url, req_options)
        pagination_item = PaginationItem.from_response(server_response.content, self.parent_srv.namespace)
        all_dataAlert_items = DataAlertItem.from_response(server_response.content, self.parent_srv.namespace)
        return all_dataAlert_items, pagination_item

    # Get 1 dataAlert
    @api(version="3.2")
    def get_by_id(self, dataAlert_id: str) -> DataAlertItem:
        if not dataAlert_id:
            error = "dataAlert ID undefined."
            raise ValueError(error)
        logger.info(f"Querying single dataAlert (ID: {dataAlert_id})")
        url = f"{self.baseurl}/{dataAlert_id}"
        server_response = self.get_request(url)
        return DataAlertItem.from_response(server_response.content, self.parent_srv.namespace)[0]

    @api(version="3.2")
    def delete(self, dataAlert: Union[DataAlertItem, str]) -> None:
        if isinstance(dataAlert, DataAlertItem):
            dataAlert_id = dataAlert.id
        elif isinstance(dataAlert, str):
            dataAlert_id = dataAlert
        else:
            raise TypeError("dataAlert should be a DataAlertItem or a string of an id.")
        if not dataAlert_id:
            error = "Dataalert ID undefined."
            raise ValueError(error)
        # DELETE /api/api-version/sites/site-id/dataAlerts/data-alert-id/users/user-id
        url = f"{self.baseurl}/{dataAlert_id}"
        self.delete_request(url)
        logger.info(f"Deleted single dataAlert (ID: {dataAlert_id})")

    @api(version="3.2")
    def delete_user_from_alert(self, dataAlert: Union[DataAlertItem, str], user: Union[UserItem, str]) -> None:
        if isinstance(dataAlert, DataAlertItem):
            dataAlert_id = dataAlert.id
        elif isinstance(dataAlert, str):
            dataAlert_id = dataAlert
        else:
            raise TypeError("dataAlert should be a DataAlertItem or a string of an id.")
        if isinstance(user, UserItem):
            user_id = user.id
        elif isinstance(user, str):
            user_id = user
        else:
            raise TypeError("user should be a UserItem or a string of an id.")
        if not dataAlert_id:
            error = "Dataalert ID undefined."
            raise ValueError(error)
        if not user_id:
            error = "User ID undefined."
            raise ValueError(error)
        # DELETE /api/api-version/sites/site-id/dataAlerts/data-alert-id/users/user-id
        url = f"{self.baseurl}/{dataAlert_id}/users/{user_id}"
        self.delete_request(url)
        logger.info(f"Deleted User (ID {user_id}) from dataAlert (ID: {dataAlert_id})")

    @api(version="3.2")
    def add_user_to_alert(self, dataAlert_item: DataAlertItem, user: Union[UserItem, str]) -> UserItem:
        if isinstance(user, UserItem):
            user_id = user.id
        elif isinstance(user, str):
            user_id = user
        else:
            raise TypeError("user should be a UserItem or a string of an id.")
        if not dataAlert_item.id:
            error = "Dataalert item missing ID."
            raise MissingRequiredFieldError(error)
        if not user_id:
            error = "User ID undefined."
            raise ValueError(error)
        url = f"{self.baseurl}/{dataAlert_item.id}/users"
        update_req = RequestFactory.DataAlert.add_user_to_alert(dataAlert_item, user_id)
        server_response = self.post_request(url, update_req)
        logger.info(f"Added user (ID {user_id}) to dataAlert item (ID: {dataAlert_item.id})")
        added_user = UserItem.from_response(server_response.content, self.parent_srv.namespace)[0]
        return added_user

    @api(version="3.2")
    def update(self, dataAlert_item: DataAlertItem) -> DataAlertItem:
        if not dataAlert_item.id:
            error = "Dataalert item missing ID."
            raise MissingRequiredFieldError(error)

        url = f"{self.baseurl}/{dataAlert_item.id}"
        update_req = RequestFactory.DataAlert.update_req(dataAlert_item)
        server_response = self.put_request(url, update_req)
        logger.info(f"Updated dataAlert item (ID: {dataAlert_item.id})")
        updated_dataAlert = DataAlertItem.from_response(server_response.content, self.parent_srv.namespace)[0]
        return updated_dataAlert