def get_view_query()

in cid/common.py [0:0]

• 34 lines of code
• 18 McCabe index (conditional complexity)

    def get_view_query(self, view_name: str) -> str:
        """ Returns a fully compiled AHQ """
        # View path
        view_definition = self.resources.get('views').get(view_name, dict())
        cur_required = view_definition.get('dependsOn', dict()).get('cur')
        if cur_required and self.cur.hasSavingsPlans and self.cur.hasReservations and view_definition.get('spriFile'):
            view_file = view_definition.get('spriFile')
        elif cur_required and self.cur.hasSavingsPlans and view_definition.get('spFile'):
            view_file = view_definition.get('spFile')
        elif cur_required and self.cur.hasReservations and view_definition.get('riFile'):
            view_file = view_definition.get('riFile')
        elif view_definition.get('File'):
            view_file = view_definition.get('File')
        else:
            logger.critical(
                f'\n"{view_name}" view information is incorrect, skipping')

        # Load TPL file
        template = Template(resource_string(view_definition.get(
            'providedBy'), f'data/queries/{view_file}').decode('utf-8'))

        # Prepare template parameters
        columns_tpl = dict()
        columns_tpl.update({
            'cur_table_name': self.cur.tableName if cur_required else None,
            'athena_database_name': self.athena.DatabaseName if view_definition.get('parameters', dict()).get('athenaDatabaseName') else None
        })
        for k,v in view_definition.get('parameters', dict()).items():
            if k == 'athenaDatabaseName':
                param = {'athena_database_name': self.athena.DatabaseName}
            elif v.get('value'):
                param = {k:v.get('value')}
            else:
                value = None
                while not value:
                    value = click.prompt(f"Required parameter: {k} ({v.get('description')})", default=v.get('value'), show_default=True)
                param = {k:value}
            # Add parameter
            columns_tpl.update(param)
        # Compile template
        compiled_query = template.safe_substitute(columns_tpl)

        return compiled_query