def create_view()

in cid/common.py [0:0]

• 29 lines of code
• 8 McCabe index (conditional complexity)

    def create_view(self, view_name: str) -> None:
        # For account mappings create a view using a special helper
        if view_name in ['account_map', 'aws_accounts']:
            self.accountMap.create(view_name)
            return
        # Create a view
        print(f'\nCreating view: {view_name}')
        logger.info(f'Creating view: {view_name}')
        logger.info(f'Getting view definition')
        view_definition = self.resources.get('views').get(view_name, dict())
        logger.debug(f'View definition: {view_definition}')
        # Discover dependency views (may not be discovered earlier)
        dependency_views = view_definition.get(
            'dependsOn', dict()).get('views', list())
        logger.info(f"Dependency views: {', '.join(dependency_views)}" if dependency_views else 'No dependency views')
        self.athena.discover_views(dependency_views)
        while dependency_views:
            dep = dependency_views.copy().pop()
        # for dep in dependency_views:
            if dep not in self.athena._metadata.keys():
                print(f'Missing dependency view: {dep}, trying to create')
                logger.info(f'Missing dependency view: {dep}, trying to create')
                self.create_view(dep)
            dependency_views.remove(dep)
        view_query = self.get_view_query(view_name=view_name)
        if view_definition.get('type') == 'Glue_Table':
            try:
                self.glue.create_table(json.loads(view_query))
            except self.glue.client.exceptions.AlreadyExistsException:
                print(f'\nError: Glue table "{view_name}" exists but not found, please check your configuration, exiting')
                exit(1)
        else:
            self.athena.execute_query(view_query)