python类import_string()的实例源码

def _create_cache(backend, **kwargs):
    try:
        # Try to get the CACHES entry for the given backend name first
        try:
            conf = settings.CACHES[backend]
        except KeyError:
            try:
                # Trying to import the given backend, in case it's a dotted path
                import_string(backend)
            except ImportError as e:
                raise InvalidCacheBackendError("Could not find backend '%s': %s" % (
                    backend, e))
            location = kwargs.pop('LOCATION', '')
            params = kwargs
        else:
            params = conf.copy()
            params.update(kwargs)
            backend = params.pop('BACKEND')
            location = params.pop('LOCATION', '')
        backend_cls = import_string(backend)
    except ImportError as e:
        raise InvalidCacheBackendError(
            "Could not find backend '%s': %s" % (backend, e))
    return backend_cls(location, params)

def find_template_loader(self, loader):
        if isinstance(loader, (tuple, list)):
            args = list(loader[1:])
            loader = loader[0]
        else:
            args = []

        if isinstance(loader, six.string_types):
            loader_class = import_string(loader)

            if getattr(loader_class, '_accepts_engine_in_init', False):
                args.insert(0, self)
            else:
                warnings.warn(
                    "%s inherits from django.template.loader.BaseLoader "
                    "instead of django.template.loaders.base.Loader. " %
                    loader, RemovedInDjango110Warning, stacklevel=2)

            return loader_class(*args)
        else:
            raise ImproperlyConfigured(
                "Invalid value in template loaders configuration: %r" % loader)

def __getitem__(self, alias):
        try:
            return self._engines[alias]
        except KeyError:
            try:
                params = self.templates[alias]
            except KeyError:
                raise InvalidTemplateEngineError(
                    "Could not find config for '{}' "
                    "in settings.TEMPLATES".format(alias))

            # If importing or initializing the backend raises an exception,
            # self._engines[alias] isn't set and this code may get executed
            # again, so we must preserve the original params. See #24265.
            params = params.copy()
            backend = params.pop('BACKEND')
            engine_cls = import_string(backend)
            engine = engine_cls(params)

            self._engines[alias] = engine
            return engine

def configure_logging(logging_config, logging_settings):
    if not sys.warnoptions:
        # Route warnings through python logging
        logging.captureWarnings(True)
        # RemovedInNextVersionWarning is a subclass of DeprecationWarning which
        # is hidden by default, hence we force the "default" behavior
        warnings.simplefilter("default", RemovedInNextVersionWarning)

    if logging_config:
        # First find the logging configuration function ...
        logging_config_func = import_string(logging_config)

        logging.config.dictConfig(DEFAULT_LOGGING)

        # ... then invoke it with the logging settings
        if logging_settings:
            logging_config_func(logging_settings)

def import_backend(dotted_path):
    """
    Theres two formats for the dotted_path.
    One with the backend class (old) and one without (new)
    eg:
      old: wagtail.wagtailsearch.backends.elasticsearch.ElasticsearchSearchBackend
      new: wagtail.wagtailsearch.backends.elasticsearch

    If a new style dotted path was specified, this function would
    look for a backend class from the "SearchBackend" attribute.
    """
    try:
        # New
        backend_module = import_module(dotted_path)
        return backend_module.SearchBackend
    except ImportError as e:
        try:
            # Old
            return import_string(dotted_path)
        except ImportError:
            six.reraise(ImportError, e, sys.exc_info()[2])

def _create_cache(backend, **kwargs):
    try:
        # Try to get the CACHES entry for the given backend name first
        try:
            conf = settings.CACHES[backend]
        except KeyError:
            try:
                # Trying to import the given backend, in case it's a dotted path
                import_string(backend)
            except ImportError as e:
                raise InvalidCacheBackendError("Could not find backend '%s': %s" % (
                    backend, e))
            location = kwargs.pop('LOCATION', '')
            params = kwargs
        else:
            params = conf.copy()
            params.update(kwargs)
            backend = params.pop('BACKEND')
            location = params.pop('LOCATION', '')
        backend_cls = import_string(backend)
    except ImportError as e:
        raise InvalidCacheBackendError(
            "Could not find backend '%s': %s" % (backend, e))
    return backend_cls(location, params)

def get_callback_function(setting_name, default=None):
    func = getattr(settings, setting_name, None)

    if not func:
        return default

    if callable(func):
        return func

    if isinstance(func, six.string_types):
        func = import_string(func)

    if not callable(func):
        raise ImproperlyConfigured(
            '{name} must be callable.'.format(name=setting_name)
        )

    return func

def _create_cache(backend, **kwargs):
    try:
        # Try to get the CACHES entry for the given backend name first
        try:
            conf = settings.CACHES[backend]
        except KeyError:
            try:
                # Trying to import the given backend, in case it's a dotted path
                import_string(backend)
            except ImportError as e:
                raise InvalidCacheBackendError("Could not find backend '%s': %s" % (
                    backend, e))
            location = kwargs.pop('LOCATION', '')
            params = kwargs
        else:
            params = conf.copy()
            params.update(kwargs)
            backend = params.pop('BACKEND')
            location = params.pop('LOCATION', '')
        backend_cls = import_string(backend)
    except ImportError as e:
        raise InvalidCacheBackendError(
            "Could not find backend '%s': %s" % (backend, e))
    return backend_cls(location, params)

def __init__(self, sitekey=None, secretkey=None, timeout=None, pass_on_error=None, **kwargs):
        """
        :param sitekey: site key (public)
        :param secretkey: secret key (private)
        :param timeout: connection to recaptcha service timeout
        :param pass_on_error: do not raise exception if recaptcha service is not working.
        """
        self.sitekey = sitekey or getattr(settings, 'RECAPTCHA_SITEKEY')
        self.secretkey = secretkey or getattr(settings, 'RECAPTCHA_SECRETKEY')

        if timeout is None:
            timeout = getattr(settings, 'RECAPTCHA_TIMEOUT', default_settings.RECAPTCHA_TIMEOUT)
        self.timeout = timeout

        if pass_on_error is None:
            pass_on_error = getattr(settings, 'RECAPTCHA_PASS_ON_ERROR', default_settings.RECAPTCHA_PASS_ON_ERROR)
        self.pass_on_error = pass_on_error

        if not 'widget' in kwargs:
            recaptcha_widget = import_string(getattr(settings, 'RECAPTCHA_WIDGET', default_settings.RECAPTCHA_WIDGET))
            kwargs['widget'] = recaptcha_widget(sitekey=self.sitekey)
        elif isinstance(kwargs['widget'], type):
            kwargs['widget'] = kwargs['widget'](sitekey=self.sitekey)

        super().__init__(**kwargs)

def init(run_woker=True):
    from django.utils.module_loading import import_string
    global _runners
    configs = getattr(settings, "EASY_JOB", {})
    logger = logging.getLogger(configs.get('logger', None))
    runners = {}
    for worker_name, worker_options in configs['workers'].items():
        # noinspection PyPep8Naming
        try:
            if 'result_backend' not in worker_options or worker_options['result_backend'] is None:
                worker_options['result_backend'] = "easy_job.result_backends.dummy.DummyBackend"
            WorkerInitializerClass = import_string(worker_options.pop('initializer'))
            worker = WorkerInitializerClass(**worker_options)
            runners[worker_name] = worker.start(not run_woker)
        except (AttributeError, ValueError):
            logger.fatal("invalid initializer specified for worker with name {}".format(worker_name))
    if len(runners) == 0:
        logger.warning("No worker is available")
    _runners = runners

def get_backend(service_name):
    """
    Gets the backend with the given service settings.
    It checks for a BACKEND key and later passes in these OPTIONS in the
    backend.

    :param service_settings: dict like object
    :returns: base.BaseBackend inherited object
    """
    backend_import = DEFAULT_SERVICE
    service_settings = settings.USER_VERIFICATION.get(service_name, None)

    if service_settings is None:
        raise ValueError("service with {} key not found".format(service_name))

    if service_settings.get('BACKEND', None):
        backend_import = service_settings.get('BACKEND', None)

    backend_cls = import_string(backend_import)
    return backend_cls(identifier=service_name, **service_settings.get('OPTIONS', {}))

def run_check(id, request=None, fail_silently=True, fail_status=500):
    status = 200
    try:
        v = config.checks[id]
        if isinstance(v, six.string_types):
            c = import_string(v)
            ret, status = c(request)
        elif callable(v):
            ret, status = v(request)
        else:
            ret = v
    except Exception as e:
        ret = "ERROR"
        status = fail_status
        logger.exception(e)
        if settings.DEBUG:
            ret = str(e)
        if not fail_silently:
            raise

    return ret, status

def get_extra(config, request=None):
    extras = {}
    for k, v in config.extra.items():
        try:
            if isinstance(v, six.string_types):
                c = import_string(v)
                extras[k] = c(request)
            elif callable(v):
                extras[k] = v(request)
            else:
                extras[k] = v
        except Exception as e:
            logger.exception(e)
            if settings.DEBUG:
                extras[k] = str(e)
    return extras

def _create_cache(backend, **kwargs):
    try:
        # Try to get the CACHES entry for the given backend name first
        try:
            conf = settings.CACHES[backend]
        except KeyError:
            try:
                # Trying to import the given backend, in case it's a dotted path
                import_string(backend)
            except ImportError as e:
                raise InvalidCacheBackendError("Could not find backend '%s': %s" % (
                    backend, e))
            location = kwargs.pop('LOCATION', '')
            params = kwargs
        else:
            params = conf.copy()
            params.update(kwargs)
            backend = params.pop('BACKEND')
            location = params.pop('LOCATION', '')
        backend_cls = import_string(backend)
    except ImportError as e:
        raise InvalidCacheBackendError(
            "Could not find backend '%s': %s" % (backend, e))
    return backend_cls(location, params)

def _create_cache(backend, **kwargs):
    try:
        # Try to get the CACHES entry for the given backend name first
        try:
            conf = settings.CACHES[backend]
        except KeyError:
            try:
                # Trying to import the given backend, in case it's a dotted path
                import_string(backend)
            except ImportError as e:
                raise InvalidCacheBackendError("Could not find backend '%s': %s" % (
                    backend, e))
            location = kwargs.pop('LOCATION', '')
            params = kwargs
        else:
            params = conf.copy()
            params.update(kwargs)
            backend = params.pop('BACKEND')
            location = params.pop('LOCATION', '')
        backend_cls = import_string(backend)
    except ImportError as e:
        raise InvalidCacheBackendError(
            "Could not find backend '%s': %s" % (backend, e))
    return backend_cls(location, params)

def _create_cache(backend, **kwargs):
    try:
        # Try to get the CACHES entry for the given backend name first
        try:
            conf = settings.CACHES[backend]
        except KeyError:
            try:
                # Trying to import the given backend, in case it's a dotted path
                import_string(backend)
            except ImportError as e:
                raise InvalidCacheBackendError("Could not find backend '%s': %s" % (
                    backend, e))
            location = kwargs.pop('LOCATION', '')
            params = kwargs
        else:
            params = conf.copy()
            params.update(kwargs)
            backend = params.pop('BACKEND')
            location = params.pop('LOCATION', '')
        backend_cls = import_string(backend)
    except ImportError as e:
        raise InvalidCacheBackendError(
            "Could not find backend '%s': %s" % (backend, e))
    return backend_cls(location, params)

def _create_cache(backend, **kwargs):
    try:
        # Try to get the CACHES entry for the given backend name first
        try:
            conf = settings.CACHES[backend]
        except KeyError:
            try:
                # Trying to import the given backend, in case it's a dotted path
                import_string(backend)
            except ImportError as e:
                raise InvalidCacheBackendError("Could not find backend '%s': %s" % (
                    backend, e))
            location = kwargs.pop('LOCATION', '')
            params = kwargs
        else:
            params = conf.copy()
            params.update(kwargs)
            backend = params.pop('BACKEND')
            location = params.pop('LOCATION', '')
        backend_cls = import_string(backend)
    except ImportError as e:
        raise InvalidCacheBackendError(
            "Could not find backend '%s': %s" % (backend, e))
    return backend_cls(location, params)

def _create_cache(backend, **kwargs):
    try:
        # Try to get the CACHES entry for the given backend name first
        try:
            conf = settings.CACHES[backend]
        except KeyError:
            try:
                # Trying to import the given backend, in case it's a dotted path
                import_string(backend)
            except ImportError as e:
                raise InvalidCacheBackendError("Could not find backend '%s': %s" % (
                    backend, e))
            location = kwargs.pop('LOCATION', '')
            params = kwargs
        else:
            params = conf.copy()
            params.update(kwargs)
            backend = params.pop('BACKEND')
            location = params.pop('LOCATION', '')
        backend_cls = import_string(backend)
    except ImportError as e:
        raise InvalidCacheBackendError(
            "Could not find backend '%s': %s" % (backend, e))
    return backend_cls(location, params)

def _to_cls(self, dotted_path):
        return import_string(dotted_path)

def get_offset_backend():
    default = 'logpipe.backend.kafka.ModelOffsetStore'
    backend_path = settings.get('OFFSET_BACKEND', default)
    return import_string(backend_path)()

def get_consumer_backend(topic_name, **kwargs):
    default = 'logpipe.backend.kafka.Consumer'
    backend_path = settings.get('CONSUMER_BACKEND', default)
    return import_string(backend_path)(topic_name, **kwargs)

def get_producer_backend():
    default = 'logpipe.backend.kafka.Producer'
    backend_path = settings.get('PRODUCER_BACKEND', default)
    return import_string(backend_path)()

def usage_for_periods(*args, **kwargs):
    name = getattr(settings, 'SOUVENIRS_USAGE_REPORTS_FUNCTION', None)
    func = import_string(name) if name else _usage_for_periods
    return func(*args, **kwargs)

def wamp_dispatch(message):
    func = import_string(message.content['func_path'])
    return func(*message.content['args'], **message.content['kwargs'])

def call_processors(self, purpose, **kwargs):
        for proc in self.get_processors(purpose):
            if isinstance(proc, str):
                proc = getattr(self, proc, None) or import_string(proc)
            kwargs['purpose'] = purpose
            proc(**kwargs)

def load_handler(path, *args, **kwargs):
    """
    Given a path to a handler, return an instance of that handler.

    E.g.::
        >>> from django.http import HttpRequest
        >>> request = HttpRequest()
        >>> load_handler('django.core.files.uploadhandler.TemporaryFileUploadHandler', request)
        <TemporaryFileUploadHandler object at 0x...>
    """
    return import_string(path)(*args, **kwargs)

def get_storage_class(import_path=None):
    return import_string(import_path or settings.DEFAULT_FILE_STORAGE)

def get_cookie_signer(salt='django.core.signing.get_cookie_signer'):
    Signer = import_string(settings.SIGNING_BACKEND)
    key = force_bytes(settings.SECRET_KEY)
    return Signer(b'django.http.cookies' + key, salt=salt)

def load_middleware(self):
        """
        Populate middleware lists from settings.MIDDLEWARE_CLASSES.

        Must be called after the environment is fixed (see __call__ in subclasses).
        """
        self._view_middleware = []
        self._template_response_middleware = []
        self._response_middleware = []
        self._exception_middleware = []

        request_middleware = []
        for middleware_path in settings.MIDDLEWARE_CLASSES:
            mw_class = import_string(middleware_path)
            try:
                mw_instance = mw_class()
            except MiddlewareNotUsed as exc:
                if settings.DEBUG:
                    if six.text_type(exc):
                        logger.debug('MiddlewareNotUsed(%r): %s', middleware_path, exc)
                    else:
                        logger.debug('MiddlewareNotUsed: %r', middleware_path)
                continue

            if hasattr(mw_instance, 'process_request'):
                request_middleware.append(mw_instance.process_request)
            if hasattr(mw_instance, 'process_view'):
                self._view_middleware.append(mw_instance.process_view)
            if hasattr(mw_instance, 'process_template_response'):
                self._template_response_middleware.insert(0, mw_instance.process_template_response)
            if hasattr(mw_instance, 'process_response'):
                self._response_middleware.insert(0, mw_instance.process_response)
            if hasattr(mw_instance, 'process_exception'):
                self._exception_middleware.insert(0, mw_instance.process_exception)

        # We only assign to this when initialization is complete as it is used
        # as a flag for initialization being complete.
        self._request_middleware = request_middleware

def get_internal_wsgi_application():
    """
    Loads and returns the WSGI application as configured by the user in
    ``settings.WSGI_APPLICATION``. With the default ``startproject`` layout,
    this will be the ``application`` object in ``projectname/wsgi.py``.

    This function, and the ``WSGI_APPLICATION`` setting itself, are only useful
    for Django's internal server (runserver); external WSGI servers should just
    be configured to point to the correct application object directly.

    If settings.WSGI_APPLICATION is not set (is ``None``), we just return
    whatever ``django.core.wsgi.get_wsgi_application`` returns.
    """
    from django.conf import settings
    app_path = getattr(settings, 'WSGI_APPLICATION')
    if app_path is None:
        return get_wsgi_application()

    try:
        return import_string(app_path)
    except ImportError as e:
        msg = (
            "WSGI application '%(app_path)s' could not be loaded; "
            "Error importing module: '%(exception)s'" % ({
                'app_path': app_path,
                'exception': e,
            })
        )
        six.reraise(ImproperlyConfigured, ImproperlyConfigured(msg),
                    sys.exc_info()[2])