python类resource_stream()的实例源码

 复制代码def schematron(cls, schema):
        transforms = [
            "xml/schematron/iso_dsdl_include.xsl",
            "xml/schematron/iso_abstract_expand.xsl",
            "xml/schematron/iso_svrl_for_xslt1.xsl",
            ]
        if isinstance(schema, file):
            compiled = etree.parse(schema)
        else:
            compiled = schema
        for filename in transforms:
            with resource_stream(
                    __name__, filename) as stream:
                xform_xml = etree.parse(stream)
                xform = etree.XSLT(xform_xml)
                compiled = xform(compiled)
        return etree.XSLT(compiled)

 复制代码def __init__(self, cli_args):
        """Load config from commandline arguments"""
        self.cli_args = cli_args
        self.set_logging()

        # Don't even bother loading the config file if it's just a version query
        if cli_args.version:
            print("Dactyl version %s" % __version__)
            exit(0)

        self.bypass_errors = cli_args.bypass_errors

        # Start with the default config, then overwrite later
        self.config = yaml.load(resource_stream(__name__, "default-config.yml"))
        self.filters = {}
        if cli_args.config:
            self.load_config_from_file(cli_args.config)
        else:
            logger.debug("No config file specified, trying ./dactyl-config.yml")
            self.load_config_from_file(DEFAULT_CONFIG_FILE)
        self.load_filters()

 复制代码def get_es_template(self, filename):
        """Loads an ElasticSearch template (as JSON)"""
        template_path = os.path.join(self.config["template_path"], filename)
        try:
            with open(template_path) as f:
                es_template = json.load(f)
        except (FileNotFoundError, json.decoder.JSONDecodeError) as e:
            if type(e) == FileNotFoundError:
                logger.debug("Didn't find ES template (%s), falling back to default" %
                    template_path)
            elif type(e) == json.decoder.JSONDecodeError:
                recoverable_error(("Error JSON-decoding ES template (%s)" %
                    template_path), self.bypass_errors)
            with resource_stream(__name__, BUILTIN_ES_TEMPLATE) as f:
                es_template = json.load(f)
        return es_template

 复制代码def __open_resource(name):
    """Open a resource from the zoneinfo subdir for reading.

    Uses the pkg_resources module if available and no standard file
    found at the calculated location.
    """
    name_parts = name.lstrip('/').split('/')
    for part in name_parts:
        if part == os.path.pardir or os.path.sep in part:
            raise ValueError('Bad path segment: %r' % part)
    filename = os.path.join(os.path.dirname(__file__),
                            'zoneinfo', *name_parts)
    if not os.path.exists(filename) and resource_stream is not None:
        # http://bugs.launchpad.net/bugs/383171 - we avoid using this
        # unless absolutely necessary to help when a broken version of
        # pkg_resources is installed.
        return resource_stream(__name__, 'zoneinfo/' + name)
    return open(filename, 'rb')

 复制代码def load_model(package_name, model_dir, model_name):
    model_path = model_dir + "/" + model_name
    model = {}
    for f in pkg_resources.resource_listdir(package_name, model_path):
        f = model_path + '/' + f
        with pkg_resources.resource_stream(package_name, f) as fd:
            append_model(model, yaml.safe_load(fd))
    imports_path = model.get('imports')
    if imports_path:
        f = model_dir + '/' + imports_path
        with pkg_resources.resource_stream(package_name, f) as fd:
            append_model(model, yaml.safe_load(fd))
    extend_base_objects(model)
    extend_api_objects(model)
    return model


# Singleton generator

 复制代码def _open_asset_path(path, encoding=None):
    """
    :param asset_path:
        string containing absolute path to file,
        or package-relative path using format
        ``"python.module:relative/file/path"``.

    :returns:
        filehandle opened in 'rb' mode
        (unless encoding explicitly specified)
    """
    if encoding:
        return codecs.getreader(encoding)(_open_asset_path(path))
    if os.path.isabs(path):
        return open(path, "rb")
    package, sep, subpath = path.partition(":")
    if not sep:
        raise ValueError("asset path must be absolute file path "
                         "or use 'pkg.name:sub/path' format: %r" % (path,))
    return pkg_resources.resource_stream(package, subpath)


#: type aliases

 复制代码def open_resource(name):
    """Open a resource from the zoneinfo subdir for reading.

    Uses the pkg_resources module if available and no standard file
    found at the calculated location.
    """
    name_parts = name.lstrip('/').split('/')
    for part in name_parts:
        if part == os.path.pardir or os.path.sep in part:
            raise ValueError('Bad path segment: %r' % part)
    filename = os.path.join(os.path.dirname(__file__),
                            'zoneinfo', *name_parts)
    if not os.path.exists(filename) and resource_stream is not None:
        # http://bugs.launchpad.net/bugs/383171 - we avoid using this
        # unless absolutely necessary to help when a broken version of
        # pkg_resources is installed.
        return resource_stream(__name__, 'zoneinfo/' + name)
    return open(filename, 'rb')

 复制代码def prepare_packer_template(config, template_name):
    """
    Prepares a packer template JSON file according to configuration and writes
    it into a temporary location where packer later expects it.

    Uses jinja2 template syntax to generate the resulting JSON file.
    Templates are in templates/ and snippets in templates/snippets/.
    """
    try:
        template_fd = resource_stream(__name__,
                                     'templates/{}.json'.format(template_name))
    except FileNotFoundError:
        print("Template doesn't exist: {}".format(template_name))
        sys.exit(2)

    filepath = resource_filename(__name__, 'templates/')
    env = Environment(loader=FileSystemLoader(filepath), autoescape=False,
                      trim_blocks=True, lstrip_blocks=True)
    template = env.get_template("{}.json".format(template_name))

    # write to temporary file
    f = create_cachefd('{}.json'.format(template_name))
    f.write(template.render(config)) # pylint: disable=no-member
    f.close()
    return f.name

 复制代码def open_resource(self, name):
        """Open a resource from the zoneinfo subdir for reading.

        Uses the pkg_resources module if available and no standard file
        found at the calculated location.
        """
        name_parts = name.lstrip('/').split('/')
        for part in name_parts:
            if part == os.path.pardir or os.path.sep in part:
                raise ValueError('Bad path segment: %r' % part)
        filename = os.path.join(os.path.dirname(__file__),
                                'zoneinfo', *name_parts)
        if not os.path.exists(filename) and resource_stream is not None:
            # http://bugs.launchpad.net/bugs/383171 - we avoid using this
            # unless absolutely necessary to help when a broken version of
            # pkg_resources is installed.
            return resource_stream(__name__, 'zoneinfo/' + name)
        return open(filename, 'rb')

 复制代码def schema_from_json(path):
    """ Create a pyspark schema from the json representation.

    The json representation must be from a StructType. This can be
    generated from any StructType using the `.json()` method. The
    schema for a dataframe can be obtained using the `.schema`
    accessor. For example, to generate the json from the
    `topline_summary`, run the following in the pyspark repl:

    >>> path = 's3a://telemetry-parquet/topline_summary/v1/mode=weekly'
    >>> json_data = spark.read.parquet(path).schema.json()

    :path str: Path the the json data
    """
    with pkg_resources.resource_stream(mozetl.topline.__name__, path) as f:
        data = json.load(f)
        return StructType.fromJson(data)


# Generate module level schemas

 复制代码def _transform_to_html(self, content, xslt_package=None, xslt_path=None):

        xslt_package = xslt_package or __name__
        xslt_path = xslt_path or \
            '../templates/ckanext/spatial/gemini2-html-stylesheet.xsl'

        # optimise -- read transform only once and compile rather
        # than at each request
        with resource_stream(xslt_package, xslt_path) as style:
            style_xml = etree.parse(style)
            transformer = etree.XSLT(style_xml)

        xml = etree.parse(StringIO(content.encode('utf-8')))
        html = transformer(xml)

        response.headers['Content-Type'] = 'text/html; charset=utf-8'
        response.headers['Content-Length'] = len(content)

        result = etree.tostring(html, pretty_print=True)

        return result

 复制代码def open_resource(name):
    """Open a resource from the zoneinfo subdir for reading.

    Uses the pkg_resources module if available and no standard file
    found at the calculated location.
    """
    name_parts = name.lstrip('/').split('/')
    for part in name_parts:
        if part == os.path.pardir or os.path.sep in part:
            raise ValueError('Bad path segment: %r' % part)
    filename = os.path.join(os.path.dirname(__file__),
                            'zoneinfo', *name_parts)
    if not os.path.exists(filename) and resource_stream is not None:
        # http://bugs.launchpad.net/bugs/383171 - we avoid using this
        # unless absolutely necessary to help when a broken version of
        # pkg_resources is installed.
        return resource_stream(__name__, 'zoneinfo/' + name)
    return open(filename, 'rb')

 复制代码def open_resource(name):
    """Open a resource from the zoneinfo subdir for reading.

    Uses the pkg_resources module if available and no standard file
    found at the calculated location.
    """
    name_parts = name.lstrip('/').split('/')
    for part in name_parts:
        if part == os.path.pardir or os.path.sep in part:
            raise ValueError('Bad path segment: %r' % part)
    filename = os.path.join(os.path.dirname(__file__),
                            'zoneinfo', *name_parts)
    if not os.path.exists(filename):
        # http://bugs.launchpad.net/bugs/383171 - we avoid using this
        # unless absolutely necessary to help when a broken version of
        # pkg_resources is installed.
        try:
            from pkg_resources import resource_stream
        except ImportError:
            resource_stream = None

        if resource_stream is not None:
            return resource_stream(__name__, 'zoneinfo/' + name)
    return open(filename, 'rb')

 复制代码def open_resource(name):
    """Open a resource from the zoneinfo subdir for reading.

    Uses the pkg_resources module if available and no standard file
    found at the calculated location.
    """
    name_parts = name.lstrip('/').split('/')
    for part in name_parts:
        if part == os.path.pardir or os.path.sep in part:
            raise ValueError('Bad path segment: %r' % part)
    filename = os.path.join(os.path.dirname(__file__),
                            'zoneinfo', *name_parts)
    if not os.path.exists(filename):
        # http://bugs.launchpad.net/bugs/383171 - we avoid using this
        # unless absolutely necessary to help when a broken version of
        # pkg_resources is installed.
        try:
            from pkg_resources import resource_stream
        except ImportError:
            resource_stream = None

        if resource_stream is not None:
            return resource_stream(__name__, 'zoneinfo/' + name)
    return open(filename, 'rb')

 复制代码def open_resource(name):
    """Open a resource from the zoneinfo subdir for reading.

    Uses the pkg_resources module if available and no standard file
    found at the calculated location.
    """
    name_parts = name.lstrip('/').split('/')
    for part in name_parts:
        if part == os.path.pardir or os.path.sep in part:
            raise ValueError('Bad path segment: %r' % part)
    filename = os.path.join(os.path.dirname(__file__),
                            'zoneinfo', *name_parts)
    if not os.path.exists(filename):
        # http://bugs.launchpad.net/bugs/383171 - we avoid using this
        # unless absolutely necessary to help when a broken version of
        # pkg_resources is installed.
        try:
            from pkg_resources import resource_stream
        except ImportError:
            resource_stream = None

        if resource_stream is not None:
            return resource_stream(__name__, 'zoneinfo/' + name)
    return open(filename, 'rb')

 复制代码def open_resource(name):
    """Open a resource from the zoneinfo subdir for reading.

    Uses the pkg_resources module if available and no standard file
    found at the calculated location.
    """
    name_parts = name.lstrip('/').split('/')
    for part in name_parts:
        if part == os.path.pardir or os.path.sep in part:
            raise ValueError('Bad path segment: %r' % part)
    filename = os.path.join(os.path.dirname(__file__),
                            'zoneinfo', *name_parts)
    if not os.path.exists(filename) and resource_stream is not None:
        # http://bugs.launchpad.net/bugs/383171 - we avoid using this
        # unless absolutely necessary to help when a broken version of
        # pkg_resources is installed.
        return resource_stream(__name__, 'zoneinfo/' + name)
    return open(filename, 'rb')

 复制代码def create_blank_config(conda_fp, project_fp, template="default"):

    if template=="microb120":
        template_stream = pkg_resources.resource_stream(
            "sunbeamlib", "data/microb120_config.yml")
    elif template=="microb191":
        template_stream = pkg_resources.resource_stream(
            "sunbeamlib", "data/microb191_config.yml")
    elif template=="pmacs":
        template_stream = pkg_resources.resource_stream(
            "sunbeamlib", "data/pmacs_config.yml")
    elif template=="respublica":
        template_stream = pkg_resources.resource_stream(
            "sunbeamlib", "data/default_config.yml")
    else:
        sys.stderr.write("Using default config template...\n")
        template_stream = pkg_resources.resource_stream(
            "sunbeamlib", "data/default_config.yml")

    return template_stream.read().decode().format(
        CONDA_FP=conda_fp, PROJECT_FP=project_fp)

 复制代码def open_resource(name):
    """Open a resource from the zoneinfo subdir for reading.

    Uses the pkg_resources module if available and no standard file
    found at the calculated location.
    """
    name_parts = name.lstrip('/').split('/')
    for part in name_parts:
        if part == os.path.pardir or os.path.sep in part:
            raise ValueError('Bad path segment: %r' % part)
    filename = os.path.join(os.path.dirname(__file__),
                            'zoneinfo', *name_parts)
    if not os.path.exists(filename) and resource_stream is not None:
        # http://bugs.launchpad.net/bugs/383171 - we avoid using this
        # unless absolutely necessary to help when a broken version of
        # pkg_resources is installed.
        return resource_stream(__name__, 'zoneinfo/' + name)
    return open(filename, 'rb')

 复制代码def open_resource(name):
    """Open a resource from the zoneinfo subdir for reading.

    Uses the pkg_resources module if available and no standard file
    found at the calculated location.
    """
    name_parts = name.lstrip('/').split('/')
    for part in name_parts:
        if part == os.path.pardir or os.path.sep in part:
            raise ValueError('Bad path segment: %r' % part)
    filename = os.path.join(os.path.dirname(__file__),
                            'zoneinfo', *name_parts)
    if not os.path.exists(filename) and resource_stream is not None:
        # http://bugs.launchpad.net/bugs/383171 - we avoid using this
        # unless absolutely necessary to help when a broken version of
        # pkg_resources is installed.
        return resource_stream(__name__, 'zoneinfo/' + name)
    return open(filename, 'rb')

 复制代码def __open_resource(name):
    """Open a resource from the zoneinfo subdir for reading.

    Uses the pkg_resources module if available and no standard file
    found at the calculated location.
    """
    name_parts = name.lstrip('/').split('/')
    for part in name_parts:
        if part == os.path.pardir or os.path.sep in part:
            raise ValueError('Bad path segment: %r' % part)
    filename = os.path.join(os.path.dirname(__file__),
                            'zoneinfo', *name_parts)
    if not os.path.exists(filename) and resource_stream is not None:
        # http://bugs.launchpad.net/bugs/383171 - we avoid using this
        # unless absolutely necessary to help when a broken version of
        # pkg_resources is installed.
        return resource_stream(__name__, 'zoneinfo/' + name)
    return open(filename, 'rb')

 复制代码def open_resource(name):
    """Open a resource from the zoneinfo subdir for reading.

    Uses the pkg_resources module if available and no standard file
    found at the calculated location.
    """
    name_parts = name.lstrip('/').split('/')
    for part in name_parts:
        if part == os.path.pardir or os.path.sep in part:
            raise ValueError('Bad path segment: %r' % part)
    filename = os.path.join(os.path.dirname(__file__),
                            'zoneinfo', *name_parts)
    if not os.path.exists(filename):
        # http://bugs.launchpad.net/bugs/383171 - we avoid using this
        # unless absolutely necessary to help when a broken version of
        # pkg_resources is installed.
        try:
            from pkg_resources import resource_stream
        except ImportError:
            resource_stream = None

        if resource_stream is not None:
            return resource_stream(__name__, 'zoneinfo/' + name)
    return open(filename, 'rb')

 复制代码def open_resource(name):
    """Open a resource from the zoneinfo subdir for reading.

    Uses the pkg_resources module if available and no standard file
    found at the calculated location.
    """
    name_parts = name.lstrip('/').split('/')
    for part in name_parts:
        if part == os.path.pardir or os.path.sep in part:
            raise ValueError('Bad path segment: %r' % part)
    filename = os.path.join(os.path.dirname(__file__),
                            'zoneinfo', *name_parts)
    if not os.path.exists(filename) and resource_stream is not None:
        # http://bugs.launchpad.net/bugs/383171 - we avoid using this
        # unless absolutely necessary to help when a broken version of
        # pkg_resources is installed.
        return resource_stream(__name__, 'zoneinfo/' + name)
    return open(filename, 'rb')

 复制代码def read_from_egg(tfile):
    '''Read a relative path, getting the contents
    locally or from the installed egg, parsing the contents
    based on file_type if given, such as yaml
    Params:
        tfile: relative package path
        file_type: file extension such as "json" or "yaml" or None

    Returns:
        contents: yaml or json loaded or raw
    '''
    template_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), tfile)
    if not os.path.exists(template_path):
        path_in_egg = os.path.join("btax", tfile)
        buf = resource_stream(Requirement.parse("btax"), path_in_egg)
        _bytes = buf.read()
        contents = str(_bytes)
    else:
        with open(template_path, 'r') as f:
            contents = f.read()
    return contents

 复制代码def open_resource(name):
    """Open a resource from the zoneinfo subdir for reading.

    Uses the pkg_resources module if available and no standard file
    found at the calculated location.
    """
    name_parts = name.lstrip('/').split('/')
    for part in name_parts:
        if part == os.path.pardir or os.path.sep in part:
            raise ValueError('Bad path segment: %r' % part)
    filename = os.path.join(os.path.dirname(__file__),
                            'zoneinfo', *name_parts)
    if not os.path.exists(filename):
        # http://bugs.launchpad.net/bugs/383171 - we avoid using this
        # unless absolutely necessary to help when a broken version of
        # pkg_resources is installed.
        try:
            from pkg_resources import resource_stream
        except ImportError:
            resource_stream = None

        if resource_stream is not None:
            return resource_stream(__name__, 'zoneinfo/' + name)
    return open(filename, 'rb')

 复制代码def open_resource(name):
    """Open a resource from the zoneinfo subdir for reading.

    Uses the pkg_resources module if available and no standard file
    found at the calculated location.
    """
    name_parts = name.lstrip('/').split('/')
    for part in name_parts:
        if part == os.path.pardir or os.path.sep in part:
            raise ValueError('Bad path segment: %r' % part)
    filename = os.path.join(os.path.dirname(__file__),
                            'zoneinfo', *name_parts)
    if not os.path.exists(filename):
        # http://bugs.launchpad.net/bugs/383171 - we avoid using this
        # unless absolutely necessary to help when a broken version of
        # pkg_resources is installed.
        try:
            from pkg_resources import resource_stream
        except ImportError:
            resource_stream = None

        if resource_stream is not None:
            return resource_stream(__name__, 'zoneinfo/' + name)
    return open(filename, 'rb')

 复制代码def test_resource_api(easter_fixture):
    test_file = NamedTemporaryFile(mode='wb+')
    dirname, file_name = os.path.split(test_file.name)

    easter_fixture.stub_egg.location = dirname
    easter_fixture.stub_egg.activate()

    assert pkg_resources.resource_exists(easter_fixture.stub_egg.as_requirement(), file_name)
    assert not pkg_resources.resource_exists(easter_fixture.stub_egg.as_requirement(), 'IDoNotExist')

    contents = b'asdd '
    test_file.write(contents)
    test_file.flush()

    as_string = pkg_resources.resource_string(easter_fixture.stub_egg.as_requirement(), file_name)
    assert as_string == contents

    as_file = pkg_resources.resource_stream(easter_fixture.stub_egg.as_requirement(), file_name)
    assert as_file.read() == contents

 复制代码def open_resource(name):
    """Open a resource from the zoneinfo subdir for reading.

    Uses the pkg_resources module if available and no standard file
    found at the calculated location.
    """
    name_parts = name.lstrip('/').split('/')
    for part in name_parts:
        if part == os.path.pardir or os.path.sep in part:
            raise ValueError('Bad path segment: %r' % part)
    filename = os.path.join(os.path.dirname(__file__),
                            'zoneinfo', *name_parts)
    if not os.path.exists(filename) and resource_stream is not None:
        # http://bugs.launchpad.net/bugs/383171 - we avoid using this
        # unless absolutely necessary to help when a broken version of
        # pkg_resources is installed.
        return resource_stream(__name__, 'zoneinfo/' + name)
    return open(filename, 'rb')

 复制代码def open_resource(name):
    """Open a resource from the zoneinfo subdir for reading.

    Uses the pkg_resources module if available and no standard file
    found at the calculated location.
    """
    name_parts = name.lstrip('/').split('/')
    for part in name_parts:
        if part == os.path.pardir or os.path.sep in part:
            raise ValueError('Bad path segment: %r' % part)
    filename = os.path.join(os.path.dirname(__file__),
                            'zoneinfo', *name_parts)
    if not os.path.exists(filename) and resource_stream is not None:
        # http://bugs.launchpad.net/bugs/383171 - we avoid using this
        # unless absolutely necessary to help when a broken version of
        # pkg_resources is installed.
        return resource_stream(__name__, 'zoneinfo/' + name)
    return open(filename, 'rb')

 复制代码def open_resource(name):
    """Open a resource from the zoneinfo subdir for reading.

    Uses the pkg_resources module if available and no standard file
    found at the calculated location.
    """
    name_parts = name.lstrip('/').split('/')
    for part in name_parts:
        if part == os.path.pardir or os.path.sep in part:
            raise ValueError('Bad path segment: %r' % part)
    filename = os.path.join(os.path.dirname(__file__),
                            'zoneinfo', *name_parts)
    if not os.path.exists(filename) and resource_stream is not None:
        # http://bugs.launchpad.net/bugs/383171 - we avoid using this
        # unless absolutely necessary to help when a broken version of
        # pkg_resources is installed.
        return resource_stream(__name__, 'zoneinfo/' + name)
    return open(filename, 'rb')

 复制代码def open_resource(name):
    """Open a resource from the zoneinfo subdir for reading.

    Uses the pkg_resources module if available and no standard file
    found at the calculated location.
    """
    name_parts = name.lstrip('/').split('/')
    for part in name_parts:
        if part == os.path.pardir or os.path.sep in part:
            raise ValueError('Bad path segment: %r' % part)
    filename = os.path.join(os.path.dirname(__file__),
                            'zoneinfo', *name_parts)
    if not os.path.exists(filename) and resource_stream is not None:
        # http://bugs.launchpad.net/bugs/383171 - we avoid using this
        # unless absolutely necessary to help when a broken version of
        # pkg_resources is installed.
        return resource_stream(__name__, 'zoneinfo/' + name)
    return open(filename, 'rb')