-
Notifications
You must be signed in to change notification settings - Fork 7
/
load_template_google_sheets.py
296 lines (226 loc) · 11.4 KB
/
load_template_google_sheets.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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
"""
This application demonstrates how to create a Tag Template in Data Catalog,
loading its information from Google Sheets.
"""
import argparse
import logging
import re
import stringcase
import unicodedata
from google.api_core import exceptions
from google.cloud import datacatalog
from googleapiclient import discovery
from googleapiclient import errors
from oauth2client import service_account
_CLOUD_PLATFORM_REGION = 'us-central1'
_CUSTOM_MULTIVALUED_TYPE = 'MULTI'
_DATA_CATALOG_BOOL_TYPE = 'BOOL'
_DATA_CATALOG_ENUM_TYPE = 'ENUM'
_DATA_CATALOG_NATIVE_TYPES = ['BOOL', 'DOUBLE', 'ENUM', 'STRING', 'TIMESTAMP']
_LOOKING_FOR_SHEET_LOG_FORMAT = 'Looking for {} sheet {} | {}...'
class TemplateMaker:
def __init__(self):
self.__sheets_reader = GoogleSheetsReader()
self.__datacatalog_facade = DataCatalogFacade()
def run(self, spreadsheet_id, project_id, template_id, display_name, delete_existing=False):
master_template_fields = self.__sheets_reader.read_master(
spreadsheet_id, stringcase.spinalcase(template_id))
self.__process_native_fields(spreadsheet_id, project_id, template_id, display_name,
master_template_fields, delete_existing)
self.__process_custom_multivalued_fields(spreadsheet_id, project_id, template_id,
display_name, master_template_fields,
delete_existing)
def __process_native_fields(self, spreadsheet_id, project_id, template_id, display_name,
master_template_fields, delete_existing_template):
native_fields = self.__filter_fields_by_types(master_template_fields,
_DATA_CATALOG_NATIVE_TYPES)
StringFormatter.format_elements_to_snakecase(native_fields, 0)
enums_names = {}
for field in native_fields:
if not field[2] == _DATA_CATALOG_ENUM_TYPE:
continue
names_from_sheet = self.__sheets_reader.read_helper(spreadsheet_id,
stringcase.spinalcase(field[0]))
enums_names[field[0]] = [name[0] for name in names_from_sheet]
template_name = datacatalog.DataCatalogClient.tag_template_path(
project_id, _CLOUD_PLATFORM_REGION, template_id)
if delete_existing_template:
self.__datacatalog_facade.delete_tag_template(template_name)
if not self.__datacatalog_facade.tag_template_exists(template_name):
self.__datacatalog_facade.create_tag_template(project_id, template_id, display_name,
native_fields, enums_names)
def __process_custom_multivalued_fields(self, spreadsheet_id, project_id, template_id,
display_name, master_template_fields,
delete_existing_template):
multivalued_fields = self.__filter_fields_by_types(master_template_fields,
[_CUSTOM_MULTIVALUED_TYPE])
StringFormatter.format_elements_to_snakecase(multivalued_fields, 0)
for field in multivalued_fields:
try:
values_from_sheet = self.__sheets_reader.read_helper(
spreadsheet_id, stringcase.spinalcase(field[0]))
fields = [(StringFormatter.format_to_snakecase(value[0]), value[0],
_DATA_CATALOG_BOOL_TYPE) for value in values_from_sheet]
except errors.HttpError as err:
if err.resp.status in [400]:
logging.info('NOT FOUND. Ignoring...')
continue # Ignore creating a new template representing the multivalued field
else:
raise
custom_template_id = f'{template_id}_{field[0]}'
custom_display_name = f'{display_name} - {field[1]}'
template_name = datacatalog.DataCatalogClient.tag_template_path(
project_id, _CLOUD_PLATFORM_REGION, custom_template_id)
if delete_existing_template:
self.__datacatalog_facade.delete_tag_template(template_name)
if not self.__datacatalog_facade.tag_template_exists(template_name):
self.__datacatalog_facade.create_tag_template(project_id, custom_template_id,
custom_display_name, fields)
@classmethod
def __filter_fields_by_types(cls, fields, valid_types):
return [field for field in fields if field[2] in valid_types]
"""
Input reader
========================================
"""
class GoogleSheetsReader:
def __init__(self):
self.__sheets_facade = GoogleSheetsFacade()
def read_master(self, spreadsheet_id, sheet_name, values_per_line=3):
return self.__read(spreadsheet_id, sheet_name, 'master', values_per_line)
def read_helper(self, spreadsheet_id, sheet_name, values_per_line=1):
return self.__read(spreadsheet_id, sheet_name, 'helper', values_per_line)
def __read(self, spreadsheet_id, sheet_name, sheet_type, values_per_line):
"""
Read the requested values from each line and store them into a list.
:param spreadsheet_id: Spreadsheet ID.
:param sheet_name: Sheet name.
:param sheet_type: Sheet type {'master', 'helper'}.
:param values_per_line: Number of consecutive values to be read from each line.
"""
logging.info(_LOOKING_FOR_SHEET_LOG_FORMAT.format(sheet_type, spreadsheet_id, sheet_name))
sheet_data = self.__sheets_facade.read_sheet(spreadsheet_id, sheet_name, values_per_line)
data = []
logging.info(f'Reading spreadsheet {spreadsheet_id} | {sheet_name}...')
for row in sheet_data.get('valueRanges')[0].get('values'):
row_data = []
for counter in range(values_per_line):
row_data.append(row[counter].strip())
data.append(row_data)
# The first line is usually used for headers, so it's discarded.
del (data[0])
logging.info('DONE')
return data
"""
API communication classes
========================================
"""
class DataCatalogFacade:
"""
Manage Templates by communicating to Data Catalog's API.
"""
def __init__(self):
# Initialize the API client.
self.__datacatalog = datacatalog.DataCatalogClient()
def create_tag_template(self,
project_id,
template_id,
display_name,
fields_descriptors,
enums_names=None):
"""Create a Tag Template."""
location = datacatalog.DataCatalogClient.common_location_path(
project_id, _CLOUD_PLATFORM_REGION)
tag_template = datacatalog.TagTemplate()
tag_template.display_name = display_name
for descriptor in fields_descriptors:
field = datacatalog.TagTemplateField()
field.display_name = descriptor[1]
field_id = descriptor[0]
field_type = descriptor[2]
if not field_type == _DATA_CATALOG_ENUM_TYPE:
field.type_.primitive_type = datacatalog.FieldType.PrimitiveType[field_type]
else:
for enum_name in enums_names[field_id]:
enum_value = datacatalog.FieldType.EnumType.EnumValue()
enum_value.display_name = enum_name
field.type_.enum_type.allowed_values.append(enum_value)
tag_template.fields[field_id] = field
created_tag_template = self.__datacatalog.create_tag_template(parent=location,
tag_template_id=template_id,
tag_template=tag_template)
logging.info(f'===> Template created: {created_tag_template.name}')
def delete_tag_template(self, name):
"""Delete a Tag Template."""
try:
self.__datacatalog.delete_tag_template(name=name, force=True)
logging.info(f'===> Template deleted: {name}')
except exceptions.PermissionDenied:
pass
def tag_template_exists(self, name):
"""Check if a Tag Template with the provided name already exists."""
try:
self.__datacatalog.get_tag_template(name=name)
return True
except exceptions.PermissionDenied:
return False
class GoogleSheetsFacade:
"""
Access spreadsheets data by communicating to the Google Sheets API.
"""
def __init__(self):
# Initialize the API client.
self.__service = discovery.build(
serviceName='sheets',
version='v4',
credentials=service_account.ServiceAccountCredentials.get_application_default(),
cache_discovery=False)
def read_sheet(self, spreadsheet_id, sheet_name, values_per_line):
return self.__service.spreadsheets().values().batchGet(
spreadsheetId=spreadsheet_id,
ranges=f'{sheet_name}!A:{chr(ord("@") + values_per_line)}').execute()
"""
Tools & utilities
========================================
"""
class StringFormatter:
@classmethod
def format_elements_to_snakecase(cls, a_list, internal_index=None):
if internal_index is None:
for counter in range(len(a_list)):
a_list[counter] = cls.format_to_snakecase(a_list[counter])
else:
for element in a_list:
element[internal_index] = cls.format_to_snakecase(element[internal_index])
@classmethod
def format_to_snakecase(cls, string):
normalized_str = unicodedata.normalize('NFKD', string).encode('ASCII', 'ignore').decode()
normalized_str = re.sub(r'[^a-zA-Z0-9]+', ' ', normalized_str)
normalized_str = normalized_str.strip()
normalized_str = normalized_str.lower() \
if (' ' in normalized_str) or (normalized_str.isupper()) \
else stringcase.camelcase(normalized_str) # FooBarBaz => fooBarBaz
return stringcase.snakecase(normalized_str) # foo-bar-baz => foo_bar_baz
"""
Main program entry point
========================================
"""
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
logging.getLogger('googleapiclient.discovery').setLevel(logging.ERROR)
logging.getLogger('oauth2client.client').setLevel(logging.ERROR)
logging.getLogger('oauth2client.transport').setLevel(logging.ERROR)
parser = argparse.ArgumentParser(description='Load Tag Template from CSV')
parser.add_argument('--template-id', help='the template ID', required=True)
parser.add_argument('--display-name', help='template\'s Display Name', required=True)
parser.add_argument('--project-id',
help='GCP Project in which the Template will be created',
required=True)
parser.add_argument('--spreadsheet-id', help='Google Spreadsheet ID', required=True)
parser.add_argument(
'--delete-existing',
action='store_true',
help='delete existing Templates and recreate them with the provided metadata')
args = parser.parse_args()
TemplateMaker().run(args.spreadsheet_id, args.project_id, args.template_id, args.display_name,
args.delete_existing)