-
Notifications
You must be signed in to change notification settings - Fork 0
/
workspace_copier.py
153 lines (114 loc) · 6.71 KB
/
workspace_copier.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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
import copy
from gooddata_sdk import GoodDataSdk
class CheckInputs:
def __init__(self, original_source_SDK: GoodDataSdk,
target_source_SDK: GoodDataSdk, workspaces_to_copy) -> None:
self.original_source_SDK = original_source_SDK
self.target_source_SDK = target_source_SDK
self.original_list_workspaces = self.original_source_SDK.catalog_workspace.list_workspaces()
self.workspaces_to_copy = workspaces_to_copy
def valid_workspaces(self) -> bool:
workspace_ids_original_host = [w.id for w in self.original_list_workspaces]
check = all(item in workspace_ids_original_host for item in self.workspaces_to_copy)
return True if check == True else False
def _get_data_sources(self) -> list:
data_sources = []
for workspace in self.workspaces_to_copy:
ldm = self.original_source_SDK.catalog_workspace_content.get_declarative_ldm(workspace).to_dict()
for i in ldm['ldm']['datasets']:
if i['dataSourceTableId']['dataSourceId'] not in data_sources:
data_sources.append(i['dataSourceTableId']['dataSourceId'])
return data_sources
def data_sources_duplicated_by_id(self) -> bool:
original_data_sources = self._get_data_sources()
target_data_sources = [i.id for i in self.target_source_SDK.catalog_data_source.list_data_sources()]
check = any(item in original_data_sources for item in target_data_sources)
if check:
print("Abort data sources transition! Check your destinatation list of data sources. Target Instance has Data source with the same ID!")
return True
else:
print("No duplicates are found")
return False
@property
def data_sources(self):
return self._get_data_sources()
class WorkspacesProcurement:
def __init__(self, original_source_SDK: GoodDataSdk, target_source_SDK: GoodDataSdk) -> None:
self.original_source_SDK = original_source_SDK
self.target_source_SDK = target_source_SDK
self.original_list_workspaces = self.original_source_SDK.catalog_workspace.list_workspaces()
self.collected_parents_workspaces = []
self.declarative_workspace_filters = self.original_source_SDK.catalog_workspace.get_declarative_workspace_data_filters().to_dict()
self.workspaceDataFilters = list(self.declarative_workspace_filters["workspaceDataFilters"])
self.target_list_workspaces = self.target_source_SDK.catalog_workspace.list_workspaces()
self.workspace_ids_target_host = [w.id for w in self.target_list_workspaces]
def workspace_info(self, workspace_id) -> dict:
w = self.original_source_SDK._catalog_workspace.get_workspace(workspace_id)
return {
'id': workspace_id,
'name': w.name,
'parent': w.parent_id
}
def create_workspace(self, CatalogWorkspace, workspace_id, prefix: str, postfix: str) -> str or None:
self.info = self.workspace_info(workspace_id)
self.name = self.info['name']
self.parent = self.info['parent']
if prefix+workspace_id+postfix not in self.workspace_ids_target_host:
# if parent exists - do not touch it
if self.parent is not None:
self.parent = prefix+self.parent+postfix
else:
self.parent = ''
self.collected_parents_workspaces.append(workspace_id)
self.target_source_SDK.catalog_workspace.create_or_update(
CatalogWorkspace(
workspace_id=prefix + workspace_id + postfix,
name=prefix + self.name + postfix,
parent_id=self.parent
)
)
return workspace_id
else:
print(workspace_id, "-> with added prefix has been already in the target host. Skipping it's creation ")
print("Collected Parent Workspaces ->", self.collected_parents_workspaces)
def _get_parent(self, workspace_id: str, instance: None or str = None) -> str:
if instance == 'target':
self.workspace = self.target_source_SDK.catalog_workspace.get_workspace(workspace_id)
if instance is None or instance == 'original':
self.workspace = self.original_source_SDK.catalog_workspace.get_workspace(workspace_id)
return self.workspace.parent_id
def restore_hierarchy(self, workspace_id: str) -> list:
self.catalog = []
self.catalog.append(workspace_id)
self.parent = self._get_parent(workspace_id)
while self.parent is not None:
self.catalog.append(self.parent)
self.parent = self._get_parent(self.parent)
return self.catalog
def get_and_load_LDM_and_ADM(self, from_workspace_id: str, prefix: str, postfix: str) -> None:
self.ldm_to_load, self.adm_to_load = '', ''
self.ldm_to_load = self.original_source_SDK.catalog_workspace_content.get_declarative_ldm(from_workspace_id)
self.adm_to_load = self.original_source_SDK.catalog_workspace_content.get_declarative_analytics_model(from_workspace_id)
self.target_source_SDK.catalog_workspace_content.put_declarative_ldm(prefix + from_workspace_id + postfix, self.ldm_to_load)
self.target_source_SDK.catalog_workspace_content.put_declarative_analytics_model(prefix + from_workspace_id + postfix, self.adm_to_load)
def extract_data_filters(self, workspaces: list, prefix: str, postfix: str) -> list:
self.output = []
for dataFilter in self.workspaceDataFilters:
dataFilter_copy = copy.deepcopy(dataFilter)
dataFilter_copy['workspaceDataFilterSettings'] = []
for filterValues in dataFilter['workspaceDataFilterSettings']:
if filterValues.get('workspace')['id'] in workspaces:
filterValues_copy = copy.deepcopy(filterValues)
filterValues_copy['id'] = prefix + filterValues_copy['id'] + postfix
filterValues_copy['workspace']['id'] = prefix + filterValues_copy['workspace']['id'] + postfix
dataFilter_copy['workspaceDataFilterSettings'].append(filterValues_copy)
if dataFilter_copy['workspaceDataFilterSettings'] == []:
del dataFilter_copy
else:
self.output.append(dataFilter_copy)
dataFilter_copy['id'] = prefix + dataFilter_copy['id'] + postfix
dataFilter_copy['workspace']['id'] = prefix + dataFilter_copy['workspace']['id'] + postfix
return self.output
@property
def get_parents_workspaces(self) -> list:
return self.collected_parents_workspaces