Recursive update (#243)

* Feature: recursive update
* Feature: parameters for update
* Refactoring: CUR, Dashboards, Datasets
* Feature: Athena database auto detection
* New classes: Dataset
* Fix: plugin loader
* Improvement: logging
* Bump version

Co-authored-by: Iakov Gan <[email protected]>
Co-authored-by: Oleksandr Moskalenko <[email protected]>

Author: 3 people

cid/_version.py

@@ -1,2 +1,2 @@
 
-__version__ = '0.1.22'
+__version__ = '0.1.23'

cid/cli.py

@@ -124,11 +124,12 @@ def delete(ctx, dashboard_id, **kwargs):
 
 
 @click.option('--dashboard-id', help='QuickSight dashboard id', default=None)
-@click.option('--force/--noforce', help='Allow force update', default=False)
+@click.option('--force/--noforce', help='allow selecting  up to date dashboards (flags must be before options)', default=False)
+@click.option('--recursive/--norecursive', help='Recursive update all Datasets and Views (flags must be before options)', default=False)
 @cid_command
-def update(ctx, dashboard_id, force):
+def update(ctx, dashboard_id, force, recursive, **kwargs):
     """Update Dashboard"""
-    ctx.obj.update(dashboard_id, force=force)
+    ctx.obj.update(dashboard_id, force=force, recursive=recursive)
 
 
 @click.option('--dashboard-id', help='QuickSight dashboard id', default=None)

cid/common.py

@@ -59,6 +59,10 @@ def CatalogName(self) -> str:
             logger.info(f'Using datacatalog: {self._CatalogName}')
         return self._CatalogName
 
+    @CatalogName.setter
+    def set_catalog_name(self, catalog):
+        self._CatalogName = catalog
+
     @property
     def DatabaseName(self) -> str:
         """ Check if Athena database exist """
@@ -92,6 +96,11 @@ def DatabaseName(self) -> str:
             logger.info(f'Using Athena database: {self._DatabaseName}')
         return self._DatabaseName
 
+    @DatabaseName.setter
+    def DatabaseName(self, database):
+        self._DatabaseName = database
+
+
     def list_data_catalogs(self) -> list:
         return self.client.list_data_catalogs().get('DataCatalogsSummary')
 
@@ -322,7 +331,7 @@ def wait_for_view(self, view_name: str, poll_interval=1, timeout=60) -> None:
     def delete_table(self, name: str, catalog: str=None, database: str=None):
         if get_parameter(
                 param_name=f'confirm-{name}',
-                message=f'Delete athena table {name}?',
+                message=f'Delete Athena table {name}?',
                 choices=['yes', 'no'],
                 default='no') != 'yes':
             return False
@@ -345,7 +354,7 @@ def delete_table(self, name: str, catalog: str=None, database: str=None):
     def delete_view(self, name: str, catalog: str=None, database: str=None):
         if get_parameter(
                 param_name=f'confirm-{name}',
-                message=f'Delete athena view {name}?',
+                message=f'Delete Athena view {name}?',
                 choices=['yes', 'no'],
                 default='no') != 'yes':
             return False

cid/helpers/athena.py

@@ -13,15 +13,13 @@ def __init__(self, session):
         self.client = session.client('glue', region_name=self.region)
 
 
-    def create_table(self, table: dict) -> dict:
-        """ Creates an AWS Glue table """
-        return self.client.create_table(**table)
-
-    def ensure_glue_table_created(self, view_name: str, view_query: str) -> None:
+    def create_or_upate_table(self, view_name: str, view_query: str) -> None:
+        table = json.loads(view_query)
         try:
-            self.create_table(json.loads(view_query))
-        except self.glue.client.exceptions.AlreadyExistsException:
+            self.client.create_table(**table)
+        except self.client.exceptions.AlreadyExistsException:
             logger.info(f'Glue table "{view_name}" exists')
+            self.client.update_table(**table)
 
     def delete_table(self, name, catalog, database):
         """ Delete an AWS Glue table """

cid/helpers/glue.py

@@ -17,7 +17,7 @@ class QuickSight():
     # Define defaults
     cidAccountId = '223485597511'
     _dashboards: Dict[str, Dashboard] = None
-    _datasets: Dict[str, Dataset] = {}
+    _datasets: Dict[str, Dataset] = None
     _datasources: dict() = {}
     _user: dict = None
 
@@ -93,6 +93,13 @@ def dashboards(self) -> Dict[str, Dashboard]:
             self.discover_dashboards()
         return self._dashboards
 
+    @property
+    def datasets(self) -> Dict[str, Dataset]:
+        """Returns a list of deployed dashboards"""
+        if self._datasets is None:
+            self.discover_datasets()
+        return self._datasets or {}
+
     @property
     def athena_datasources(self) -> dict:
         """Returns a list of existing athena datasources"""
@@ -109,7 +116,6 @@ def datasources(self) -> dict:
 
     def discover_dashboard(self, dashboardId: str):
         """Discover single dashboard"""
-        
         dashboard = self.describe_dashboard(DashboardId=dashboardId)
         # Look for dashboard definition by DashboardId
         _definition = next((v for v in self.supported_dashboards.values() if v['dashboardId'] == dashboard.id), None)
@@ -152,7 +158,7 @@ def _recoursive_add_view(view):
                 for dep_view in self.supported_views.get(view, {}).get('dependsOn', {}).get('views', []):
                     _recoursive_add_view(dep_view)
             for dataset_name in dashboard.datasets.keys():
-                for view in self.supported_datasets.get(dataset_name).get('dependsOn', {}).get('views', []):
+                for view in self.supported_datasets.get(dataset_name, {}).get('dependsOn', {}).get('views', []):
                     _recoursive_add_view(view)
             dashboard.views = all_views
             self._dashboards = self._dashboards or {}
@@ -267,7 +273,7 @@ def discover_data_sources(self) -> None:
                 self.describe_data_source(v.get('DataSourceId'))
         except Exception as e:
             logger.debug(e, stack_info=True)
-            for _,v in self._datasets.items():
+            for _,v in self.datasets.items():
                 for d in v.datasources:
                     logger.info(f'Discovering data source {d}')
                     self.describe_data_source(d)
@@ -517,7 +523,7 @@ def delete_dataset(self, id: str) -> bool:
                 AwsAccountId=self.account_id,
                 DataSetId=id
             )
-            self._datasets.pop(id)
+            self.datasets.pop(id)
         except self.client.exceptions.AccessDeniedException:
             logger.info('Access denied deleting dataset')
             return False
@@ -529,10 +535,24 @@ def delete_dataset(self, id: str) -> bool:
             return True
 
 
+    def get_datasets(self, id: str=None, name: str=None) -> Dataset:
+        """ get dataset that match parameters """
+        result = []
+        for dataset in self.datasets.values():
+            if id is not None and dataset.id != id:
+                continue
+            if name is not None and dataset.name != name:
+                continue
+            result.append(dataset)
+        return result
+
+
+
     def describe_dataset(self, id, timeout: int=1) -> dict:
         """ Describes an AWS QuickSight dataset """
-        if id in self._datasets:
+        if self._datasets and id in self._datasets:
             return self._datasets.get(id)
+        self._datasets = self._datasets or {}
         poll_interval = 1
         _dataset = None
         deadline = time.time() + timeout
@@ -556,6 +576,7 @@ def discover_datasets(self):
         """ Discover datasets in the account """
 
         logger.info('Discovering datasets')
+        self._datasets =  self._datasets or {}
         try:
             for dataset in self.list_data_sets():
                 try:
@@ -626,6 +647,7 @@ def create_dataset(self, definition: dict) -> str:
         dataset_id = None
         try:
             logger.info(f'Creating dataset {definition.get("Name")} ({dataset_id})')
+            logger.debug(f'Dataset definition: {definition}')
             response = self.client.create_data_set(**definition)
             dataset_id = response.get('DataSetId')
         except self.client.exceptions.ResourceExistsException:
@@ -646,6 +668,19 @@ def create_dataset(self, definition: dict) -> str:
         return dataset_id
 
 
+    def update_dataset(self, definition: dict) -> str:
+        """ Creates an AWS QuickSight dataset """
+        definition.update({'AwsAccountId': self.account_id})
+        logger.info(f'Updating dataset {definition.get("Name")}')
+
+        if "Permissions" in definition:
+            logger.info('Ignoring permissions for dataset update.')
+            del definition['Permissions']
+        response = self.client.update_data_set(**definition)
+        logger.info(f'Dataset {definition.get("Name")} is updated')
+        return True
+
+
     def create_dashboard(self, definition: dict, **kwargs) -> Dashboard:
         """ Creates an AWS QuickSight dashboard """
         DataSetReferences = list()
@@ -684,9 +719,12 @@ def create_dashboard(self, definition: dict, **kwargs) -> Dashboard:
 
         create_parameters = always_merger.merge(create_parameters, kwargs)
         try:
+            logger.info(f'Creating dashboard "{definition.get("name")}"')
+            logger.debug(create_parameters)
             create_status = self.client.create_dashboard(**create_parameters)
             logger.debug(create_status)
         except self.client.exceptions.ResourceExistsException:
+            logger.info(f'Dashboard {definition.get("name")} already exists')
             raise
         created_version = int(create_status['VersionArn'].split('/')[-1])
 
@@ -726,6 +764,7 @@ def update_dashboard(self, dashboard: Dashboard, **kwargs):
         }
 
         update_parameters = always_merger.merge(update_parameters, kwargs)
+        logger.info(f'Updating dashboard "{dashboard.name}"')
         logger.debug(f"Update parameters: {update_parameters}")
         update_status = self.client.update_dashboard(**update_parameters)
         logger.debug(update_status)
@@ -742,6 +781,7 @@ def update_dashboard(self, dashboard: Dashboard, **kwargs):
             'VersionNumber': updated_version
         }
         result = self.client.update_dashboard_published_version(**update_params)
+        logger.debug(result)
         if result['Status'] != 200:
             raise Exception(result)
 

cid/helpers/quicksight/__init__.py

@@ -5,6 +5,7 @@
 import os
 
 from cid.helpers.quicksight.resource import CidQsResource
+from cid.utils import is_unattendent_mode
 
 import logging
 
@@ -65,7 +66,7 @@ def status(self) -> str:
             elif not self.definition:
                 self._status = 'undiscovered'
             # Missing dataset
-            elif not self.datasets or (len(self.datasets) < len(self.definition.get('dependsOn').get('datasets'))):
+            elif not self.datasets or (len(set(self.datasets)) < len(set(self.definition.get('dependsOn').get('datasets')))):
                 self.status_detail = 'missing dataset(s)'
                 self._status = 'broken'
                 logger.info(f"Found datasets: {self.datasets}")
@@ -82,7 +83,12 @@ def status(self) -> str:
 
     @property
     def templateId(self) -> str:
-        return str(self.version.get('SourceEntityArn').split('/')[1])
+        if 'SourceEntityArn' not in self.version:
+            return ''
+        arn = self.version.get('SourceEntityArn')
+        if ":template/" not in arn:
+            return ''
+        return str(arn.split('/')[1])
 
     def find_local_config(self) -> Union[dict, None]:
 
@@ -136,11 +142,11 @@ def display_status(self) -> None:
             print(f"  Datasets: {', '.join(sorted(self.datasets.keys()))}")
         print('\n')
         if click.confirm('Display dashboard raw data?'):
-            print(json.dumps(self.dashboard, indent=4, sort_keys=True, default=str))
+            print(json.dumps(self.raw, indent=4, sort_keys=True, default=str))
 
     def display_url(self, url_template: str, launch: bool = False, **kwargs) -> None:
         url = url_template.format(dashboard_id=self.id, **kwargs)
         print(f"#######\n####### {self.name} is available at: " + url + "\n#######")
         _supported_env = os.environ.get('AWS_EXECUTION_ENV') not in ['CloudShell', 'AWS_Lambda']
-        if _supported_env and launch and click.confirm('Do you wish to open it in your browser?'):
+        if _supported_env and not is_unattendent_mode() and launch and click.confirm('Do you wish to open it in your browser?'):
                 click.launch(url)

cid/helpers/quicksight/dashboard.py

@@ -14,8 +14,20 @@ def id(self) -> str:
     def datasources(self) -> list:
         _datasources = []
         try:
-            for _,map in self.raw.get('PhysicalTableMap').items():
-                _datasources.append(map.get('RelationalTable').get('DataSourceArn').split('/')[-1])
+            for table_map in self.raw.get('PhysicalTableMap').values():
+                _datasources.append(table_map.get('RelationalTable').get('DataSourceArn').split('/')[-1])
         except Exception as e:
             logger.debug(e, stack_info=True)
         return _datasources
+
+    @property
+    def schemas (self) -> list:
+        schemas = []
+        try:
+            for table_map in self.raw.get('PhysicalTableMap').values():
+                schema = table_map.get('RelationalTable', {}).get('Schema', None)
+                if schema:
+                    schemas.append(schema)
+        except Exception as e:
+            logger.debug(e, stack_info=True)
+        return schemas

cid/helpers/quicksight/dataset.py

@@ -1,14 +1,22 @@
-import questionary
-import boto3
+import logging
+from collections.abc import Iterable
 
+import boto3
+import questionary
 from botocore.exceptions import NoCredentialsError, CredentialRetrievalError, NoRegionError
 
-import logging
 logger = logging.getLogger(__name__)
 
 params = {} # parameters from command line
 _all_yes = False # parameters from command line
 
+
+def intersection(a: Iterable, b: Iterable) -> Iterable:
+    return sorted(set(a).intersection(b))
+
+def difference(a: Iterable, b: Iterable) -> Iterable:
+    return sorted(list(set(a).difference(b)))
+
 def get_aws_region():
     return get_boto_session().region_name
 
@@ -61,6 +69,9 @@ def set_parameters(parameters: dict, all_yes: bool=False) -> None:
     global _all_yes
     _all_yes = all_yes
 
+def is_unattendent_mode() -> bool:
+    return _all_yes
+
 def get_parameters():
     return dict(params)