def synchronize()

in scripts/nfs-roles-management/internal/synchronization/synchronization.py [0:0]

• 56 lines of code
• 30 McCabe index (conditional complexity)

    def synchronize(self, user_ids=None, use_symlinks=False, filter_mask=FullMask.READ):
        def user_matches_criteria(test_user):
            user_name = test_user.lower()
            return user_ids is None or len(user_ids) == 0 or len([u for u in user_ids if u.lower() == user_name]) > 0

        def validate_storage(test_storage, share_mounts):
            if not test_storage.is_nfs() and test_storage.type != 'AZ' and test_storage.type != 'S3' and test_storage.type != 'GCP':
                return None
            # Skip sensitive storage for mounting
            if test_storage.sensitive == True:
                return None
            server_name = None
            storage_path = None
            storage_link_destination = None
            if test_storage.is_nfs():
                try:
                    if test_storage.path.lower().startswith('nfs://'):
                        if share_mounts[test_storage.share_mount_id].mount_type == "SMB":
                            search = re.search('([^\/]+)\/(.+)' ,test_storage.path[len('nfs://'):])
                            (server_name, storage_path) = search.group(1), search.group(2)
                        else:
                            (server_name, storage_path) = test_storage.path[len('nfs://'):].split(':')
                except ValueError:
                    pass
                except AttributeError:
                    pass
                if server_name is None or storage_path is None:
                    logging.warning('Wrong storage path: {}'.format(test_storage.path))
                    return None
                storage_path = re.sub('\/[\/]+', '/', storage_path)
                if storage_path.startswith('/'):
                    storage_path = storage_path[1:]
                storage_link_destination = os.path.join(self.__config__.nfs_root, server_name, storage_path)
            elif test_storage.type == 'AZ' or test_storage.type == 'S3' or test_storage.type == "GCP":
                storage_link_destination = os.path.join(self.__config__.nfs_root, test_storage.type, test_storage.name)

            if not os.path.exists(storage_link_destination):
                logging.warning('Storage mount not found at path: {}'.format(storage_link_destination))
                return None
            return storage_link_destination

        try:
            logging.info('Fetching storages...')
            self.__storages__ = []
            self.__users__ = []
            share_mounts = self.list_share_mounts()
            for storage in self.list_storages(filter_mask=filter_mask):
                storage.mount_source = validate_storage(storage, share_mounts)
                if storage.mount_source is not None:
                    self.__storages__.append(storage)
                    for user in storage.users.keys():
                        if user.username not in self.__users__:
                            self.__users__.append(user.username)
            logging.info('{} NFS storages fetched'.format(len(self.__storages__)))
            for user in self.__users__:
                if user_matches_criteria(user):
                    self.synchronize_user(user, use_symlinks=use_symlinks)
                    logging.info('')
        except Exception:
            logging.exception('Storages fetching has failed.')