import os
|
import pkg_resources
|
import sys
|
|
from zope.interface import implementer
|
|
from pyramid.interfaces import IPackageOverrides, PHASE1_CONFIG
|
|
from pyramid.exceptions import ConfigurationError
|
from pyramid.threadlocal import get_current_registry
|
|
from pyramid.config.actions import action_method
|
|
|
class OverrideProvider(pkg_resources.DefaultProvider):
|
def __init__(self, module):
|
pkg_resources.DefaultProvider.__init__(self, module)
|
self.module_name = module.__name__
|
|
def _get_overrides(self):
|
reg = get_current_registry()
|
overrides = reg.queryUtility(IPackageOverrides, self.module_name)
|
return overrides
|
|
def get_resource_filename(self, manager, resource_name):
|
""" Return a true filesystem path for resource_name,
|
co-ordinating the extraction with manager, if the resource
|
must be unpacked to the filesystem.
|
"""
|
overrides = self._get_overrides()
|
if overrides is not None:
|
filename = overrides.get_filename(resource_name)
|
if filename is not None:
|
return filename
|
return pkg_resources.DefaultProvider.get_resource_filename(
|
self, manager, resource_name
|
)
|
|
def get_resource_stream(self, manager, resource_name):
|
""" Return a readable file-like object for resource_name."""
|
overrides = self._get_overrides()
|
if overrides is not None:
|
stream = overrides.get_stream(resource_name)
|
if stream is not None:
|
return stream
|
return pkg_resources.DefaultProvider.get_resource_stream(
|
self, manager, resource_name
|
)
|
|
def get_resource_string(self, manager, resource_name):
|
""" Return a string containing the contents of resource_name."""
|
overrides = self._get_overrides()
|
if overrides is not None:
|
string = overrides.get_string(resource_name)
|
if string is not None:
|
return string
|
return pkg_resources.DefaultProvider.get_resource_string(
|
self, manager, resource_name
|
)
|
|
def has_resource(self, resource_name):
|
overrides = self._get_overrides()
|
if overrides is not None:
|
result = overrides.has_resource(resource_name)
|
if result is not None:
|
return result
|
return pkg_resources.DefaultProvider.has_resource(self, resource_name)
|
|
def resource_isdir(self, resource_name):
|
overrides = self._get_overrides()
|
if overrides is not None:
|
result = overrides.isdir(resource_name)
|
if result is not None:
|
return result
|
return pkg_resources.DefaultProvider.resource_isdir(
|
self, resource_name
|
)
|
|
def resource_listdir(self, resource_name):
|
overrides = self._get_overrides()
|
if overrides is not None:
|
result = overrides.listdir(resource_name)
|
if result is not None:
|
return result
|
return pkg_resources.DefaultProvider.resource_listdir(
|
self, resource_name
|
)
|
|
|
@implementer(IPackageOverrides)
|
class PackageOverrides(object):
|
# pkg_resources arg in kw args below for testing
|
def __init__(self, package, pkg_resources=pkg_resources):
|
loader = self._real_loader = getattr(package, '__loader__', None)
|
if isinstance(loader, self.__class__):
|
self._real_loader = None
|
# We register ourselves as a __loader__ *only* to support the
|
# setuptools _find_adapter adapter lookup; this class doesn't
|
# actually support the PEP 302 loader "API". This is
|
# excusable due to the following statement in the spec:
|
# ... Loader objects are not
|
# required to offer any useful functionality (any such functionality,
|
# such as the zipimport get_data() method mentioned above, is
|
# optional)...
|
# A __loader__ attribute is basically metadata, and setuptools
|
# uses it as such.
|
package.__loader__ = self
|
# we call register_loader_type for every instantiation of this
|
# class; that's OK, it's idempotent to do it more than once.
|
pkg_resources.register_loader_type(self.__class__, OverrideProvider)
|
self.overrides = []
|
self.overridden_package_name = package.__name__
|
|
def insert(self, path, source):
|
if not path or path.endswith('/'):
|
override = DirectoryOverride(path, source)
|
else:
|
override = FileOverride(path, source)
|
self.overrides.insert(0, override)
|
return override
|
|
def filtered_sources(self, resource_name):
|
for override in self.overrides:
|
o = override(resource_name)
|
if o is not None:
|
yield o
|
|
def get_filename(self, resource_name):
|
for source, path in self.filtered_sources(resource_name):
|
result = source.get_filename(path)
|
if result is not None:
|
return result
|
|
def get_stream(self, resource_name):
|
for source, path in self.filtered_sources(resource_name):
|
result = source.get_stream(path)
|
if result is not None:
|
return result
|
|
def get_string(self, resource_name):
|
for source, path in self.filtered_sources(resource_name):
|
result = source.get_string(path)
|
if result is not None:
|
return result
|
|
def has_resource(self, resource_name):
|
for source, path in self.filtered_sources(resource_name):
|
if source.exists(path):
|
return True
|
|
def isdir(self, resource_name):
|
for source, path in self.filtered_sources(resource_name):
|
result = source.isdir(path)
|
if result is not None:
|
return result
|
|
def listdir(self, resource_name):
|
for source, path in self.filtered_sources(resource_name):
|
result = source.listdir(path)
|
if result is not None:
|
return result
|
|
@property
|
def real_loader(self):
|
if self._real_loader is None:
|
raise NotImplementedError()
|
return self._real_loader
|
|
def get_data(self, path):
|
""" See IPEP302Loader.
|
"""
|
return self.real_loader.get_data(path)
|
|
def is_package(self, fullname):
|
""" See IPEP302Loader.
|
"""
|
return self.real_loader.is_package(fullname)
|
|
def get_code(self, fullname):
|
""" See IPEP302Loader.
|
"""
|
return self.real_loader.get_code(fullname)
|
|
def get_source(self, fullname):
|
""" See IPEP302Loader.
|
"""
|
return self.real_loader.get_source(fullname)
|
|
|
class DirectoryOverride:
|
def __init__(self, path, source):
|
self.path = path
|
self.pathlen = len(self.path)
|
self.source = source
|
|
def __call__(self, resource_name):
|
if resource_name.startswith(self.path):
|
new_path = resource_name[self.pathlen :]
|
return self.source, new_path
|
|
|
class FileOverride:
|
def __init__(self, path, source):
|
self.path = path
|
self.source = source
|
|
def __call__(self, resource_name):
|
if resource_name == self.path:
|
return self.source, ''
|
|
|
class PackageAssetSource(object):
|
"""
|
An asset source relative to a package.
|
|
If this asset source is a file, then we expect the ``prefix`` to point
|
to the new name of the file, and the incoming ``resource_name`` will be
|
the empty string, as returned by the ``FileOverride``.
|
|
"""
|
|
def __init__(self, package, prefix):
|
self.package = package
|
if hasattr(package, '__name__'):
|
self.pkg_name = package.__name__
|
else:
|
self.pkg_name = package
|
self.prefix = prefix
|
|
def get_path(self, resource_name):
|
return '%s%s' % (self.prefix, resource_name)
|
|
def get_filename(self, resource_name):
|
path = self.get_path(resource_name)
|
if pkg_resources.resource_exists(self.pkg_name, path):
|
return pkg_resources.resource_filename(self.pkg_name, path)
|
|
def get_stream(self, resource_name):
|
path = self.get_path(resource_name)
|
if pkg_resources.resource_exists(self.pkg_name, path):
|
return pkg_resources.resource_stream(self.pkg_name, path)
|
|
def get_string(self, resource_name):
|
path = self.get_path(resource_name)
|
if pkg_resources.resource_exists(self.pkg_name, path):
|
return pkg_resources.resource_string(self.pkg_name, path)
|
|
def exists(self, resource_name):
|
path = self.get_path(resource_name)
|
if pkg_resources.resource_exists(self.pkg_name, path):
|
return True
|
|
def isdir(self, resource_name):
|
path = self.get_path(resource_name)
|
if pkg_resources.resource_exists(self.pkg_name, path):
|
return pkg_resources.resource_isdir(self.pkg_name, path)
|
|
def listdir(self, resource_name):
|
path = self.get_path(resource_name)
|
if pkg_resources.resource_exists(self.pkg_name, path):
|
return pkg_resources.resource_listdir(self.pkg_name, path)
|
|
|
class FSAssetSource(object):
|
"""
|
An asset source relative to a path in the filesystem.
|
|
"""
|
|
def __init__(self, prefix):
|
self.prefix = prefix
|
|
def get_path(self, resource_name):
|
if resource_name:
|
path = os.path.join(self.prefix, resource_name)
|
else:
|
path = self.prefix
|
return path
|
|
def get_filename(self, resource_name):
|
path = self.get_path(resource_name)
|
if os.path.exists(path):
|
return path
|
|
def get_stream(self, resource_name):
|
path = self.get_filename(resource_name)
|
if path is not None:
|
return open(path, 'rb')
|
|
def get_string(self, resource_name):
|
stream = self.get_stream(resource_name)
|
if stream is not None:
|
with stream:
|
return stream.read()
|
|
def exists(self, resource_name):
|
path = self.get_filename(resource_name)
|
if path is not None:
|
return True
|
|
def isdir(self, resource_name):
|
path = self.get_filename(resource_name)
|
if path is not None:
|
return os.path.isdir(path)
|
|
def listdir(self, resource_name):
|
path = self.get_filename(resource_name)
|
if path is not None:
|
return os.listdir(path)
|
|
|
class AssetsConfiguratorMixin(object):
|
def _override(
|
self, package, path, override_source, PackageOverrides=PackageOverrides
|
):
|
pkg_name = package.__name__
|
override = self.registry.queryUtility(IPackageOverrides, name=pkg_name)
|
if override is None:
|
override = PackageOverrides(package)
|
self.registry.registerUtility(
|
override, IPackageOverrides, name=pkg_name
|
)
|
override.insert(path, override_source)
|
|
@action_method
|
def override_asset(self, to_override, override_with, _override=None):
|
""" Add a :app:`Pyramid` asset override to the current
|
configuration state.
|
|
``to_override`` is an :term:`asset specification` to the
|
asset being overridden.
|
|
``override_with`` is an :term:`asset specification` to the
|
asset that is performing the override. This may also be an absolute
|
path.
|
|
See :ref:`assets_chapter` for more
|
information about asset overrides."""
|
if to_override == override_with:
|
raise ConfigurationError(
|
'You cannot override an asset with itself'
|
)
|
|
package = to_override
|
path = ''
|
if ':' in to_override:
|
package, path = to_override.split(':', 1)
|
|
# *_isdir = override is package or directory
|
overridden_isdir = path == '' or path.endswith('/')
|
|
if os.path.isabs(override_with):
|
override_source = FSAssetSource(override_with)
|
if not os.path.exists(override_with):
|
raise ConfigurationError(
|
'Cannot override asset with an absolute path that does '
|
'not exist'
|
)
|
override_isdir = os.path.isdir(override_with)
|
override_package = None
|
override_prefix = override_with
|
else:
|
override_package = override_with
|
override_prefix = ''
|
if ':' in override_with:
|
override_package, override_prefix = override_with.split(':', 1)
|
|
__import__(override_package)
|
to_package = sys.modules[override_package]
|
override_source = PackageAssetSource(to_package, override_prefix)
|
|
override_isdir = override_prefix == '' or override_with.endswith(
|
'/'
|
)
|
|
if overridden_isdir and (not override_isdir):
|
raise ConfigurationError(
|
'A directory cannot be overridden with a file (put a '
|
'slash at the end of override_with if necessary)'
|
)
|
|
if (not overridden_isdir) and override_isdir:
|
raise ConfigurationError(
|
'A file cannot be overridden with a directory (put a '
|
'slash at the end of to_override if necessary)'
|
)
|
|
override = _override or self._override # test jig
|
|
def register():
|
__import__(package)
|
from_package = sys.modules[package]
|
override(from_package, path, override_source)
|
|
intr = self.introspectable(
|
'asset overrides',
|
(package, override_package, path, override_prefix),
|
'%s -> %s' % (to_override, override_with),
|
'asset override',
|
)
|
intr['to_override'] = to_override
|
intr['override_with'] = override_with
|
self.action(
|
None, register, introspectables=(intr,), order=PHASE1_CONFIG
|
)
|
|
override_resource = override_asset # bw compat
|
