in cid/common.py [0:0]
def create_datasets(self, _datasets: list) -> dict:
# Check dependencies
required_datasets = sorted(_datasets)
print('\nRequired datasets: \n - {}'.format('\n - '.join(required_datasets)))
try:
print('\nDetecting existing datasets...', end='')
for dataset in self.qs.list_data_sets():
try:
self.qs.describe_dataset(dataset.get('DataSetId'))
except:
continue
except self.qs.client.exceptions.AccessDeniedException:
print('no permissions, performing full discrovery...', end='')
self.qs.dashboards
for dataset in required_datasets:
dataset_definition = self.resources.get(
'datasets').get(dataset)
finally:
print('complete')
found_datasets = sorted(
set(required_datasets).intersection([v.get('Name') for v in self.qs._datasets.values()]))
missing_datasets = sorted(
list(set(required_datasets).difference(found_datasets)))
# If we miss required datasets look in saved deployments
if len(missing_datasets):
# TODO: remove below 2 lines ?
if len(found_datasets):
print('\nFound: \n - {}'.format('\n - '.join(found_datasets)))
print('\nMissing: \n - {}'.format('\n - '.join(missing_datasets)))
# Look for previously saved deployment info
print('\nLooking in saved deployments...', end='')
saved_datasets = self.find_saved_datasets(missing_datasets)
print('{}'.format('nothing found' if not len(
saved_datasets) else 'complete'))
for k, v in saved_datasets.items():
print(f'\tfound: {k}', end='')
if len(v.keys()) > 1:
# Multiple datasets
selected = questionary.select(
f'Multiple "{k}" datasets detected, please select one',
choices=v.keys()
).ask()
self.qs._datasets.update({k: v.get(selected)})
missing_datasets.remove(k)
elif len(v.keys()):
# Single dataset
print(', using')
self.qs._datasets.update({k: next(iter(v.values()))})
missing_datasets.remove(k)
# Look by DataSetId from dataset_template file
if len(missing_datasets):
# Look for previously saved deployment info
print('\nLooking by DataSetId defined in template...', end='')
for dataset_name in missing_datasets[:]:
try:
dataset_definition = self.resources.get(
'datasets').get(dataset_name)
dataset_file = dataset_definition.get('File')
# Load TPL file
if dataset_file:
raw_template = json.loads(resource_string(dataset_definition.get(
'providedBy'), f'data/datasets/{dataset_file}').decode('utf-8'))
ds = self.qs.describe_dataset(raw_template.get('DataSetId'))
if ds.get('Name') == dataset_name:
missing_datasets.remove(dataset_name)
print(f"\n\tFound {dataset_name} as {raw_template.get('DataSetId')}")
except FileNotFoundError:
logger.info(f'File "{dataset_file}" not found')
pass
except self.qs.client.exceptions.ResourceNotFoundException:
logger.info(f'Dataset "{dataset_name}" not found')
pass
except self.qs.client.exceptions.AccessDeniedException:
logger.info(f'Access denied trying to find dataset "{dataset_name}"')
pass
except:
raise
print('complete')
# If there still datasets missing try automatic creation
if len(missing_datasets):
missing_str = ', '.join(missing_datasets)
print(
f'\nThere are still {len(missing_datasets)} datasets missing: {missing_str}')
for dataset_name in missing_datasets[:]:
print(f'Creating dataset: {dataset_name}...', end='')
try:
dataset_definition = self.resources.get(
'datasets').get(dataset_name)
except:
logger.critical(
'dashboard definition is broken, unable to proceed.')
logger.critical(
f'dataset definition not found: {dataset_name}')
raise
try:
if self.create_dataset(dataset_definition):
missing_datasets.remove(dataset_name)
print('created')
else:
print('failed')
except self.qs.client.exceptions.AccessDeniedException as AccessDeniedException:
print('unable to create, missing permissions: {}'.format(AccessDeniedException))
# Last chance to enter DataSetIds manually by user
if len(missing_datasets):
missing_str = '\n - '.join(missing_datasets)
print(
f'\nThere are still {len(missing_datasets)} datasets missing: \n - {missing_str}')
print(
f"\nCan't move forward without full list, please manually create datasets and provide DataSetIds")
# Loop over the list unless we get it empty
while len(missing_datasets):
# Make a copy and then get an item from the list
dataset_name = missing_datasets.copy().pop()
_id = click.prompt(
f'\tDataSetId/Arn for {dataset_name}', type=str)
id = _id.split('/')[-1]
try:
_dataset = self.qs.describe_dataset(id)
if _dataset.get('Name') != dataset_name:
print(f"\tFound dataset with a different name: {_dataset.get('Name')}, please provide another one")
continue
self.qs._datasets.update({dataset_name: _dataset})
missing_datasets.remove(dataset_name)
print(f'\tFound, using it')
except:
print(f"\tProvided DataSetId '{id}' can't be found\n")
continue