dbms_sanya/python/plugins/file_management/utils/file_manager.py

import importlib
from pathlib import Path


providers_loaded = False
providers = {}


def load_providers():
    global providers_loaded, providers

    for file in (Path(__file__).parent.parent / 'providers').glob('*.py'):
        if file.stem in ['errors']:
            continue
        try:
            module_ = getattr(importlib.import_module(f'file_management.providers.{file.stem}'),
                              f'{file.stem.upper()}FileManager')
            providers[file.stem] = module_
        except (ImportError, AttributeError):
            continue

    providers_loaded = True


class FileManager:

    def __init__(self, uri):
        global providers_loaded, providers

        if not providers_loaded:
            load_providers()

        self._uri = uri
        provider = providers.get(self._uri.ext()[1:].lower())
        if provider is not None:
            self._file_manager = provider(uri)
        else:
            self._file_manager = None

    def is_valid(self):
        return self._file_manager is not None

    def provider_name(self):
        if self.is_valid():
            return self._file_manager.provider_name()

        return ''

    def rename_layer(self, target_uri):
        if self.is_valid():
            self._file_manager.rename_layer(target_uri)

    def rename_database(self, target_uri):
        if self.is_valid():
            self._file_manager.rename_database(target_uri)

    def layer_exists(self, target_uri):
        if self.is_valid():
            return self._file_manager.layer_exists(target_uri)
        else:
            return False

    def database_exists(self, target_uri):
        if self.is_valid():
            return self._file_manager.database_exists(target_uri)
        else:
            return False

    def ask_overwrite_layer(self, target_uri, iface):
        if self.is_valid():
            return self._file_manager.ask_overwrite_layer(target_uri, iface)
        else:
            return True

    def ask_overwrite_database(self, target_uri, iface):
        if self.is_valid():
            return self._file_manager.ask_overwrite_database(target_uri, iface)
        else:
            return True

    def delete_layer(self, target_uri):
        if self.is_valid():
            return self._file_manager.delete_layer(target_uri)

    def delete_database(self, target_uri):
        if self.is_valid():
            return self._file_manager.delete_database(target_uri)