python类import_module()的实例源码

def plugins(fetch_handlers=None):
    if not fetch_handlers:
        fetch_handlers = FETCH_HANDLERS
    plugin_list = []
    for handler_name in fetch_handlers:
        package, classname = handler_name.rsplit('.', 1)
        try:
            handler_class = getattr(
                importlib.import_module(package),
                classname)
            plugin_list.append(handler_class())
        except NotImplementedError:
            # Skip missing plugins so that they can be ommitted from
            # installation if desired
            log("FetchHandler {} not found, skipping plugin".format(
                handler_name))
    return plugin_list

def run_migration(self, migration_file, method='upgrade'):

        if RedVersion.already_run(migration_file):
            self.logger.warn("Migration %s has already been run, skipping." % migration_file)
            return

        try:
            module = importlib.import_module('redberry.models.migrations.%s' % migration_file)

            if method == 'upgrade':
                self.logger.info("Running upgrade on %s" % migration_file)
                module.upgrade(self.db)
            else:
                self.logger.info("Running downgrade on %s" % migration_file)
                module.downgrade(self.db)

            RedVersion.store_migration(migration_file)

        except Exception, e:
            self.logger.error("Error running %s" % migration_file)
            self.logger.error(e)

def load_migration_nodes():
    nodes = {None: MigrationRouteHead()}
    for path in _migrations_dir_path.iterdir():
        if not _migration_filename_pattern.match(path.name):
            continue
        module = importlib.import_module(f'quizzler.migrations.{path.stem}')
        nodes[path.stem] = MigrationNode(path.stem, module)

    # Fill linked-list info.
    for name, node in nodes.items():
        if name is None:
            continue
        prev_name = node.module.previous
        prev_node = nodes[prev_name]
        if prev_node.next is not None:
            raise MigrationRouteConflict(
                f'{name} and {prev_node.next.name} '
                f'both specify previous = {prev_name!r}',
            )
        prev_node.next = node
        node.prev = prev_node

    return nodes

def launch(task, filename, nb_cpu, nb_gpu, use_gpu, output=None, benchmark=None, extension='', sim_same_elec=None):

    from circus.shared.parser import CircusParser
    params = CircusParser(filename)

    if task not in ['filtering', 'benchmarking']:
        params.get_data_file()

    module = importlib.import_module('circus.' + task)

    if task == 'benchmarking':
        module.main(params, nb_cpu, nb_gpu, use_gpu, output, benchmark, sim_same_elec)
    elif task in ['converting', 'merging']:
        module.main(params, nb_cpu, nb_gpu, use_gpu, extension)
    else:
        module.main(params, nb_cpu, nb_gpu, use_gpu)

def create_from_settings(cls, instance, conf):
        try:
            engine_path, _, cls_name = conf['ENGINE'].rpartition('.')
            db_name = conf['NAME']
            db_options = conf['OPTIONS'] if 'OPTIONS' in conf and conf['OPTIONS'] else dict()

            # FIX for #331. Replace utf8 by utf8mb4 in the mysql driver encoding.
            if conf['ENGINE'] == 'peewee_async.MySQLDatabase' and 'charset' in db_options and db_options['charset'] == 'utf8':
                logging.info('Forcing to use \'utf8mb4\' instead of \'utf8\' for the MySQL charset option! (Fix #331).')
                db_options['charset'] = 'utf8mb4'

            # We will try to load it so we have the validation inside this class.
            engine = getattr(importlib.import_module(engine_path), cls_name)
        except ImportError:
            raise ImproperlyConfigured('Database engine doesn\'t exist!')
        except Exception as e:
            raise ImproperlyConfigured('Database configuration isn\'t complete or engine could\'t be found!')

        return cls(engine, instance, db_name, **db_options)

def run_migration(self, name, app_label, app_module, save_migration=True):
        """
        Run + apply migration to the actual database.

        :param name: Name of migration.
        :param app_label: App label.
        :param app_module: App module path.
        :param save_migration: Save migration state?
        """
        from .models.migration import Migration
        mod = importlib.import_module('{}.migrations.{}'.format(app_module, name))

        try:
            with self.db.allow_sync():
                mod.upgrade(self.migrator)

                if save_migration:
                    Migration.create(
                        app=app_label,
                        name=name,
                        applied=True
                    )
        except Exception as e:
            logger.warning('Can\'t migrate {}.migrations.{}: {}'.format(app_module, name, str(e)))
            raise

def init_app(self, app):
        """
        Initiate app, load all signal/callbacks files. (just import, they should register with decorators).

        :param app: App instance
        :type app: pyplanet.apps.AppConfig
        """
        self._current_namespace = app.label

        # Import the signals module.
        try:
            importlib.import_module('{}.signals'.format(app.name))
        except ImportError:
            pass
        self._current_namespace = None

        # Import the callbacks module.
        try:
            importlib.import_module('{}.callbacks'.format(app.name))
        except ImportError:
            pass

def load(self):
        # Make sure we load the defaults first.
        super().load()

        # Prepare the loading.
        self.module = os.environ.get('PYPLANET_SETTINGS_MODULE')

        if not self.module:
            raise ImproperlyConfigured(
                'Settings module is not defined! Please define PYPLANET_SETTINGS_MODULE in your '
                'environment or start script.'
            )

        # Add the module itself to the configuration.
        self.settings['SETTINGS_MODULE'] = self.module

        # Load the module, put the settings into the local context.
        module = importlib.import_module(self.module)

        for setting in dir(module):
            if setting.isupper():
                self.settings[setting] = getattr(module, setting)

def plugins(fetch_handlers=None):
    if not fetch_handlers:
        fetch_handlers = FETCH_HANDLERS
    plugin_list = []
    for handler_name in fetch_handlers:
        package, classname = handler_name.rsplit('.', 1)
        try:
            handler_class = getattr(
                importlib.import_module(package),
                classname)
            plugin_list.append(handler_class())
        except NotImplementedError:
            # Skip missing plugins so that they can be ommitted from
            # installation if desired
            log("FetchHandler {} not found, skipping plugin".format(
                handler_name))
    return plugin_list

def plugins(fetch_handlers=None):
    if not fetch_handlers:
        fetch_handlers = FETCH_HANDLERS
    plugin_list = []
    for handler_name in fetch_handlers:
        package, classname = handler_name.rsplit('.', 1)
        try:
            handler_class = getattr(
                importlib.import_module(package),
                classname)
            plugin_list.append(handler_class())
        except NotImplementedError:
            # Skip missing plugins so that they can be ommitted from
            # installation if desired
            log("FetchHandler {} not found, skipping plugin".format(
                handler_name))
    return plugin_list

def _verify_api(self):
        """
        Verifies the API and loads the proper VSPK version
        """
        # Checking auth parameters
        if ('api_password' not in list(self.auth.keys()) or not self.auth['api_password']) and ('api_certificate' not in list(self.auth.keys()) or
                                                                                                'api_key' not in list(self.auth.keys()) or
                                                                                                not self.auth['api_certificate'] or not self.auth['api_key']):
            self.module.fail_json(msg='Missing api_password or api_certificate and api_key parameter in auth')

        self.api_username = self.auth['api_username']
        if 'api_password' in list(self.auth.keys()) and self.auth['api_password']:
            self.api_password = self.auth['api_password']
        if 'api_certificate' in list(self.auth.keys()) and 'api_key' in list(self.auth.keys()) and self.auth['api_certificate'] and self.auth['api_key']:
            self.api_certificate = self.auth['api_certificate']
            self.api_key = self.auth['api_key']
        self.api_enterprise = self.auth['api_enterprise']
        self.api_url = self.auth['api_url']
        self.api_version = self.auth['api_version']

        try:
            global VSPK
            VSPK = importlib.import_module('vspk.{0:s}'.format(self.api_version))
        except ImportError:
            self.module.fail_json(msg='vspk is required for this module, or the API version specified does not exist.')

def serve(cwd, app, port):
    sys.path.insert(0, cwd)

    wsgi_fqn = app.rsplit('.', 1)
    wsgi_fqn_parts = wsgi_fqn[0].rsplit('/', 1)
    if len(wsgi_fqn_parts) == 2:
        sys.path.insert(0, os.path.join(cwd, wsgi_fqn_parts[0]))
    wsgi_module = importlib.import_module(wsgi_fqn_parts[-1])
    wsgi_app = getattr(wsgi_module, wsgi_fqn[1])

    # Attempt to force Flask into debug mode
    try:
        wsgi_app.debug = True
    except:  # noqa: E722
        pass

    os.environ['IS_OFFLINE'] = 'True'

    serving.run_simple(
        'localhost', int(port), wsgi_app,
        use_debugger=True, use_reloader=True, use_evalex=True)

def _setup(module, extras):
    """Install common submodules"""

    Qt.__binding__ = module.__name__

    for name in list(_common_members) + extras:
        try:
            # print("Trying %s" % name)
            submodule = importlib.import_module(
                module.__name__ + "." + name)
        except ImportError:
            # print("Failed %s" % name)
            continue

        setattr(Qt, "_" + name, submodule)

        if name not in extras:
            # Store reference to original binding,
            # but don't store speciality modules
            # such as uic or QtUiTools
            setattr(Qt, name, _new_module(name))

def scan_for_plugins(self):
        """Plugin scanner.

        Scans for plugins in the known plugin directories. Adds them to the
        available plugins dictionary, ready to be loaded.

        Returns:
            The available plugins dictionary.
        """
        _logger.debug("Scanning for plugins.")
        for category,info in CATEGORIES.items():
            _logger.debug("Scanning category {0}".format(category))
            for module in os.listdir(os.path.join(os.path.dirname(__file__),info["directory"])):
                if module == "__init__.py" or module[-3:] != ".py":
                    continue
                _logger.debug("\tFound plugin {0}".format(module[:-3]))
                importlib.import_module("aniping.{0}.{1}".format(info["directory"], module[:-3]))
                self._available_plugins[category].append(module[:-3])
        _logger.debug("All available plugins found.")
        return self._available_plugins

def upload(filename, config):
    """Upload .su3 file with transports"""
    if "transports" in config and "enabled" in config["transports"]:
        for t in config["transports"]["enabled"].split():
            if t in config:
                tconf = config[t]
            else:
                tconf = None

            try:
                importlib.import_module("pyseeder.transports.{}".format(t)) \
                                                .run(filename, tconf)
            except ImportError:
                raise PyseederException(
                        "{} transport can't be loaded".format(t))
            except TransportException as e:
                log.error("Transport error: {}".format(e))

def email(self):
        try:
            to_email = request.args.get('to_email')

            path, attr = configuration.get('email', 'EMAIL_BACKEND').rsplit('.', 1)
            logging.info("path: " + str(path))
            logging.info("attr: " + str(attr))

            module = importlib.import_module(path)
            logging.info("module: " + str(module))

            backend = getattr(module, attr)
            backend(to_email, "Test Email", "Test Email", files=None, dryrun=False)
            flash('Email Sent')
        except Exception as e:
            flash('Failed to Send Email: ' + str(e), 'error')

        return redirect("/admin/admintools", code=302)

def __init__(self,
                 username: str,
                 password: str,
                 botModule: str,
                 botconfig: Mapping,
                 numPlayers: int,
                 variant: Variant,
                 spectators: bool,
                 gameName: str,
                 *args,
                 **kwargs) -> None:
        super().__init__(*args, **kwargs)
        self.username: str = username
        self.password: str = password
        module = importlib.import_module(botModule + '.bot')
        self.botCls: Type[Bot] = module.Bot  # type: ignore
        self.botconfig: Mapping = botconfig
        self.numPlayers: int = numPlayers
        self.variant: Variant = variant
        self.spectators: bool = spectators
        self.gameName: str = gameName
        self.conn: socketIO_client.SocketIO
        self.tablePlayers: List[str] = []
        self.readyToStart: bool = False
        self.game: Optional[Game] = None

def load_config(config_name):
    config = {}

    # Get config defaults
    for key in dir(config_defaults):
        if key.isupper():
            config[key] = getattr(config_defaults, key)

    # Merge given config with defaults
    obj = importlib.import_module(config_name)
    for key in dir(obj):
        if key in config:
            value = getattr(obj, key)
            if isinstance(config[key], dict):
                config[key].update(value)
            else:
                config[key] = value

    return config

def plugins(fetch_handlers=None):
    if not fetch_handlers:
        fetch_handlers = FETCH_HANDLERS
    plugin_list = []
    for handler_name in fetch_handlers:
        package, classname = handler_name.rsplit('.', 1)
        try:
            handler_class = getattr(
                importlib.import_module(package),
                classname)
            plugin_list.append(handler_class())
        except NotImplementedError:
            # Skip missing plugins so that they can be ommitted from
            # installation if desired
            log("FetchHandler {} not found, skipping plugin".format(
                handler_name))
    return plugin_list

def plugins(fetch_handlers=None):
    if not fetch_handlers:
        fetch_handlers = FETCH_HANDLERS
    plugin_list = []
    for handler_name in fetch_handlers:
        package, classname = handler_name.rsplit('.', 1)
        try:
            handler_class = getattr(
                importlib.import_module(package),
                classname)
            plugin_list.append(handler_class())
        except NotImplementedError:
            # Skip missing plugins so that they can be ommitted from
            # installation if desired
            log("FetchHandler {} not found, skipping plugin".format(
                handler_name))
    return plugin_list

def plugins(fetch_handlers=None):
    if not fetch_handlers:
        fetch_handlers = FETCH_HANDLERS
    plugin_list = []
    for handler_name in fetch_handlers:
        package, classname = handler_name.rsplit('.', 1)
        try:
            handler_class = getattr(
                importlib.import_module(package),
                classname)
            plugin_list.append(handler_class())
        except NotImplementedError:
            # Skip missing plugins so that they can be ommitted from
            # installation if desired
            log("FetchHandler {} not found, skipping plugin".format(
                handler_name))
    return plugin_list

def plugins(fetch_handlers=None):
    if not fetch_handlers:
        fetch_handlers = FETCH_HANDLERS
    plugin_list = []
    for handler_name in fetch_handlers:
        package, classname = handler_name.rsplit('.', 1)
        try:
            handler_class = getattr(
                importlib.import_module(package),
                classname)
            plugin_list.append(handler_class())
        except NotImplementedError:
            # Skip missing plugins so that they can be ommitted from
            # installation if desired
            log("FetchHandler {} not found, skipping plugin".format(
                handler_name))
    return plugin_list

def plugins(fetch_handlers=None):
    if not fetch_handlers:
        fetch_handlers = FETCH_HANDLERS
    plugin_list = []
    for handler_name in fetch_handlers:
        package, classname = handler_name.rsplit('.', 1)
        try:
            handler_class = getattr(
                importlib.import_module(package),
                classname)
            plugin_list.append(handler_class())
        except NotImplementedError:
            # Skip missing plugins so that they can be ommitted from
            # installation if desired
            log("FetchHandler {} not found, skipping plugin".format(
                handler_name))
    return plugin_list

def from_object(self, obj):
        """Updates the values from the given object.  An object can be of one
        of the following two types:
        -   a string: in this case the object with that name will be imported
        -   an actual object reference: that object is used directly
        :param obj: an import name or object
        """
        string_types = (str,)

        if isinstance(obj, string_types):
            # github.com/pallets/werkzeug/blob/master/werkzeug/utils.py+L399
            # obj = import_string(obj)
            obj = importlib.import_module(obj)

        for key in dir(obj):
            if key.isupper():
                self[key] = getattr(obj, key)

def _setup(module, extras):
    """Install common submodules"""

    Qt.__binding__ = module.__name__

    for name in list(_common_members) + extras:
        try:
            # print("Trying %s" % name)
            submodule = importlib.import_module(
                module.__name__ + "." + name)
        except ImportError:
            # print("Failed %s" % name)
            continue

        setattr(Qt, "_" + name, submodule)

        if name not in extras:
            # Store reference to original binding,
            # but don't store speciality modules
            # such as uic or QtUiTools
            setattr(Qt, name, _new_module(name))

def as_view(path):
    def decorator(func):
        # ..
        path_name, klass_name  = (path.split(':'))
        # ...
        @inlineCallbacks
        def wrapper(router, request, *args, **kwargs):
            # ...
            module = importlib.import_module(path_name)
            Klass = getattr(module,klass_name)
            klass = Klass(router, request,*args, **kwargs)
            # ..
            result = yield defer.maybeDeferred(klass)            
            defer.returnValue(result)
        # ..
        # _conspect_name(wrapper, klass_name)
        _conspect_name(wrapper, func.__name__)        
        _conspect_param(wrapper, func)
        _conspect_param_defaults(wrapper, func)        
        return wrapper
    return decorator

def plugins(fetch_handlers=None):
    if not fetch_handlers:
        fetch_handlers = FETCH_HANDLERS
    plugin_list = []
    for handler_name in fetch_handlers:
        package, classname = handler_name.rsplit('.', 1)
        try:
            handler_class = getattr(
                importlib.import_module(package),
                classname)
            plugin_list.append(handler_class())
        except NotImplementedError:
            # Skip missing plugins so that they can be ommitted from
            # installation if desired
            log("FetchHandler {} not found, skipping plugin".format(
                handler_name))
    return plugin_list

def set_action(self, action, *args):
        """Set action and arguments to run in triggered time

        :param action: function path as anction
        :type action: str
        :param args: function arguments
        :type args: list or tuple
        """
        if isinstance(action, (types.FunctionType, types.MethodType)):
            self.action = action
        else:
            action_module = '.'.join(action.split('.')[:-1])
            action_module = importlib.import_module(action_module)
            action = action.split('.')[-1]
            self.action = getattr(action_module, action)
        self.action_args = args

def __init__(self, states: States, request: dict = None, with_graph: bool = False):
        self.states = states
        self.state = states.attributes.state
        state_names, transitions = type(states).get_states_transitions()
        machine_cls = \
            importlib.import_module('transitions.extensions').GraphMachine if with_graph else \
            importlib.import_module('transitions').Machine
        self.machine = machine_cls(
            model=self,
            states=state_names,
            initial=states.attributes.state,
            auto_transitions=False
        )

        for transition in transitions:
            self.machine.add_transition(**transition)

def _load_project_pipelines(self):
        try:
            pipelines_module_name = settings.CML_PROJECT_PIPELINES
        except AttributeError:
            logger.info('Configure CML_PROJECT_PIPELINES in settings!')
            return
        try:
            pipelines_module = importlib.import_module(pipelines_module_name)
        except ImportError:
            return
        for item_class_name in PROCESSED_ITEMS:
            try:
                pipeline_class = getattr(pipelines_module, '{}Pipeline'.format(item_class_name))
            except AttributeError:
                continue
            self._project_pipelines[item_class_name] = pipeline_class()