-
Notifications
You must be signed in to change notification settings - Fork 421
/
databases_endpoint.py
139 lines (111 loc) · 5.66 KB
/
databases_endpoint.py
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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
import logging
from typing import Union, Iterable, Set
from tableauserverclient.server.endpoint.default_permissions_endpoint import _DefaultPermissionsEndpoint
from tableauserverclient.server.endpoint.dqw_endpoint import _DataQualityWarningEndpoint
from tableauserverclient.server.endpoint.endpoint import api, Endpoint
from tableauserverclient.server.endpoint.exceptions import MissingRequiredFieldError
from tableauserverclient.server.endpoint.permissions_endpoint import _PermissionsEndpoint
from tableauserverclient.server.endpoint.resource_tagger import TaggingMixin
from tableauserverclient.server import RequestFactory
from tableauserverclient.models import DatabaseItem, TableItem, PaginationItem, Resource
from tableauserverclient.helpers.logging import logger
class Databases(Endpoint, TaggingMixin):
def __init__(self, parent_srv):
super(Databases, self).__init__(parent_srv)
self._permissions = _PermissionsEndpoint(parent_srv, lambda: self.baseurl)
self._default_permissions = _DefaultPermissionsEndpoint(parent_srv, lambda: self.baseurl)
self._data_quality_warnings = _DataQualityWarningEndpoint(parent_srv, Resource.Database)
@property
def baseurl(self):
return "{0}/sites/{1}/databases".format(self.parent_srv.baseurl, self.parent_srv.site_id)
@api(version="3.5")
def get(self, req_options=None):
logger.info("Querying all databases 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_database_items = DatabaseItem.from_response(server_response.content, self.parent_srv.namespace)
return all_database_items, pagination_item
# Get 1 database
@api(version="3.5")
def get_by_id(self, database_id):
if not database_id:
error = "database ID undefined."
raise ValueError(error)
logger.info("Querying single database (ID: {0})".format(database_id))
url = "{0}/{1}".format(self.baseurl, database_id)
server_response = self.get_request(url)
return DatabaseItem.from_response(server_response.content, self.parent_srv.namespace)[0]
@api(version="3.5")
def delete(self, database_id):
if not database_id:
error = "Database ID undefined."
raise ValueError(error)
url = "{0}/{1}".format(self.baseurl, database_id)
self.delete_request(url)
logger.info("Deleted single database (ID: {0})".format(database_id))
@api(version="3.5")
def update(self, database_item):
if not database_item.id:
error = "Database item missing ID."
raise MissingRequiredFieldError(error)
url = "{0}/{1}".format(self.baseurl, database_item.id)
update_req = RequestFactory.Database.update_req(database_item)
server_response = self.put_request(url, update_req)
logger.info("Updated database item (ID: {0})".format(database_item.id))
updated_database = DatabaseItem.from_response(server_response.content, self.parent_srv.namespace)[0]
return updated_database
# Not Implemented Yet
@api(version="99")
def populate_tables(self, database_item):
if not database_item.id:
error = "database item missing ID. database must be retrieved from server first."
raise MissingRequiredFieldError(error)
def column_fetcher():
return self._get_tables_for_database(database_item)
database_item._set_tables(column_fetcher)
logger.info("Populated tables for database (ID: {0}".format(database_item.id))
def _get_tables_for_database(self, database_item):
url = "{0}/{1}/tables".format(self.baseurl, database_item.id)
server_response = self.get_request(url)
tables = TableItem.from_response(server_response.content, self.parent_srv.namespace)
return tables
@api(version="3.5")
def populate_permissions(self, item):
self._permissions.populate(item)
@api(version="3.5")
def update_permissions(self, item, rules):
return self._permissions.update(item, rules)
@api(version="3.5")
def delete_permission(self, item, rules):
self._permissions.delete(item, rules)
@api(version="3.5")
def populate_table_default_permissions(self, item):
self._default_permissions.populate_default_permissions(item, Resource.Table)
@api(version="3.5")
def update_table_default_permissions(self, item):
return self._default_permissions.update_default_permissions(item, Resource.Table)
@api(version="3.5")
def delete_table_default_permissions(self, item):
self._default_permissions.delete_default_permission(item, Resource.Table)
@api(version="3.5")
def populate_dqw(self, item):
self._data_quality_warnings.populate(item)
@api(version="3.5")
def update_dqw(self, item, warning):
return self._data_quality_warnings.update(item, warning)
@api(version="3.5")
def add_dqw(self, item, warning):
return self._data_quality_warnings.add(item, warning)
@api(version="3.5")
def delete_dqw(self, item):
self._data_quality_warnings.clear(item)
@api(version="3.9")
def add_tags(self, item: Union[DatabaseItem, str], tags: Iterable[str]) -> Set[str]:
return super().add_tags(item, tags)
@api(version="3.9")
def delete_tags(self, item: Union[DatabaseItem, str], tags: Iterable[str]) -> None:
super().delete_tags(item, tags)
@api(version="3.9")
def update_tags(self, item: DatabaseItem) -> None:
raise NotImplementedError("Update tags is not supported for databases.")