Django文件存储 自己定制存储系统解析

from django.core.files.storage import Storage
class MyStorage(Storage):
  ...

from django.conf import settings
from django.core.files.storage import Storage
 
class MyStorage(Storage):
  def __init__(self, option=None):
    if not option:
      option = settings.CUSTOM_STORAGE_OPTIONS
    ...

def open(self, name, mode='rb'):
  """
  Retrieves the specified file from storage.
  """
  return self._open(name, mode)
 
 
def save(self, name, content, max_length=None):
  """
  Saves new content to the file specified by name. The content should be
  a proper File object or any python file-like object, ready to be read
  from the beginning.
  """
  # Get the proper name for the file, as it will actually be saved.
  if name is None:
    name = content.name
 
  if not hasattr(content, 'chunks'):
    content = File(content, name)
 
  name = self.get_available_name(name, max_length=max_length)
  return self._save(name, content)

DEFAULT_FILE_STORAGE = 'qiniustorage.backends.QiniuStorage'

STATICFILES_STORAGE = 'qiniustorage.backends.QiniuStaticStorage'

@deconstructible
class QiniuStorage(Storage):
  """
  Qiniu Storage Service
  """
  location = ""
 
  def __init__(
      self,
      access_key=QINIU_ACCESS_KEY,
      secret_key=QINIU_SECRET_KEY,
      bucket_name=QINIU_BUCKET_NAME,
      bucket_domain=QINIU_BUCKET_DOMAIN,
      secure_url=QINIU_SECURE_URL):
 
    self.auth = Auth(access_key, secret_key)
    self.bucket_name = bucket_name
    self.bucket_domain = bucket_domain
    self.bucket_manager = BucketManager(self.auth)
    self.secure_url = secure_url
 
  def _clean_name(self, name):
    """
    Cleans the name so that Windows style paths work
    """
    # Normalize Windows style paths
    clean_name = posixpath.normpath(name).replace('\\', '/')
 
    # os.path.normpath() can strip trailing slashes so we implement
    # a workaround here.
    if name.endswith('/') and not clean_name.endswith('/'):
      # Add a trailing slash as it was stripped.
      return clean_name + '/'
    else:
      return clean_name
 
  def _normalize_name(self, name):
    """
    Normalizes the name so that paths like /path/to/ignored/../foo.txt
    work. We check to make sure that the path pointed to is not outside
    the directory specified by the LOCATION setting.
    """
 
    base_path = force_text(self.location)
    base_path = base_path.rstrip('/')
 
    final_path = urljoin(base_path.rstrip('/') + "/", name)
 
    base_path_len = len(base_path)
    if (not final_path.startswith(base_path) or
        final_path[base_path_len:base_path_len + 1] not in ('', '/')):
      raise SuspiciousOperation("Attempted access to '%s' denied." %
                   name)
    return final_path.lstrip('/')
 
  de<strong>本文来源gaodai#ma#com搞@代~码^网+</strong>f _open(self, name, mode='rb'):
    return QiniuFile(name, self, mode)
 
  def _save(self, name, content):
    cleaned_name = self._clean_name(name)
    name = self._normalize_name(cleaned_name)
 
    if hasattr(content, 'chunks'):
      content_str = b''.join(chunk for chunk in content.chunks())
    else:
      content_str = content.read()
 
    self._put_file(name, content_str)
    return cleaned_name
 
  def _put_file(self, name, content):
    token = self.auth.upload_token(self.bucket_name)
    ret, info = put_data(token, name, content)
    if ret is None or ret['key'] != name:
      raise QiniuError(info)
 
  def _read(self, name):
    return requests.get(self.url(name)).content
 
  def delete(self, name):
    name = self._normalize_name(self._clean_name(name))
    if six.PY2:
      name = name.encode('utf-8')
    ret, info = self.bucket_manager.delete(self.bucket_name, name)
 
    if ret is None or info.status_code == 612:
      raise QiniuError(info)
 
  def _file_stat(self, name, silent=False):
    name = self._normalize_name(self._clean_name(name))
    if six.PY2:
      name = name.encode('utf-8')
    ret, info = self.bucket_manager.stat(self.bucket_name, name)
    if ret is None and not silent:
      raise QiniuError(info)
    return ret
 
  def exists(self, name):
    stats = self._file_stat(name, silent=True)
    return True if stats else False
 
  def size(self, name):
    stats = self._file_stat(name)
    return stats['fsize']
 
  def modified_time(self, name):
    stats = self._file_stat(name)
    time_stamp = float(stats['putTime']) / 10000000
    return datetime.datetime.fromtimestamp(time_stamp)
 
  def listdir(self, name):
    name = self._normalize_name(self._clean_name(name))
    if name and not name.endswith('/'):
      name += '/'
 
    dirlist = bucket_lister(self.bucket_manager, self.bucket_name,
                prefix=name)
    files = []
    dirs = set()
    base_parts = name.split("/")[:-1]
    for item in dirlist:
      parts = item['key'].split("/")
      parts = parts[len(base_parts):]
      if len(parts) == 1:
        # File
        files.append(parts[0])
      elif len(parts) > 1:
        # Directory
        dirs.add(parts[0])
    return list(dirs), files
 
  def url(self, name):
    name = self._normalize_name(self._clean_name(name))
    name = filepath_to_uri(name)
    protocol = u'https://' if self.secure_url else u'http://'
    return urljoin(protocol + self.bucket_domain, name)