/usr/share/pyshared/pipeline/storage.py is in python-django-pipeline 1.2.2.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | import os
try:
from staticfiles import finders
from staticfiles.storage import CachedFilesMixin, StaticFilesStorage
except ImportError:
from django.contrib.staticfiles import finders # noqa
from django.contrib.staticfiles.storage import CachedFilesMixin, StaticFilesStorage # noqa
from django.core.exceptions import ImproperlyConfigured
from django.core.files.storage import get_storage_class
from django.utils.functional import LazyObject
from pipeline.conf import settings
class PipelineMixin(object):
def post_process(self, paths, dry_run=False, **options):
if dry_run:
return []
from pipeline.packager import Packager
packager = Packager(storage=self)
for package_name in packager.packages['css']:
package = packager.package_for('css', package_name)
output_file = packager.pack_stylesheets(package)
paths[output_file] = (self, output_file)
for package_name in packager.packages['js']:
package = packager.package_for('js', package_name)
output_file = packager.pack_javascripts(package)
paths[output_file] = (self, output_file)
super_class = super(PipelineMixin, self)
if hasattr(super_class, 'post_process'):
return super_class.post_process(paths, dry_run, **options)
return [
(path, path, True)
for path in paths
]
def get_available_name(self, name):
if self.exists(name):
self.delete(name)
return name
class PipelineStorage(PipelineMixin, StaticFilesStorage):
pass
class PipelineCachedStorage(PipelineMixin, CachedFilesMixin, StaticFilesStorage):
pass
class BaseFinderStorage(PipelineStorage):
finders = None
def __init__(self, finders=None, *args, **kwargs):
if finders is not None:
self.finders = finders
if self.finders is None:
raise ImproperlyConfigured("The storage %r doesn't have a finders class assigned." % self.__class__)
super(BaseFinderStorage, self).__init__(*args, **kwargs)
def path(self, name):
path = self.finders.find(name)
if not path:
path = super(BaseFinderStorage, self).path(name)
return path
def exists(self, name):
exists = self.finders.find(name) != None
if not exists:
return super(BaseFinderStorage, self).exists(name)
return exists
def listdir(self, path):
for finder in finders.get_finders():
for storage in finder.storages.values():
try:
return storage.listdir(path)
except OSError:
pass
def find_storage(self, name):
for finder in finders.get_finders():
for path, storage in finder.list([]):
if path.startswith(os.path.dirname(name)):
return storage
return None
def _open(self, name, mode="rb"):
storage = self.find_storage(name)
if storage:
return storage._open(name, mode)
return super(BaseFinderStorage, self)._open(name, mode)
def _save(self, name, content):
storage = self.find_storage(name)
if storage:
# Ensure we overwrite file, since we have no control on external storage
if storage.exists(name):
storage.delete(name)
return storage._save(name, content)
return super(BaseFinderStorage, self)._save(name, content)
class PipelineFinderStorage(BaseFinderStorage):
finders = finders
class DefaultStorage(LazyObject):
def _setup(self):
self._wrapped = get_storage_class(settings.PIPELINE_STORAGE)()
default_storage = DefaultStorage()
|