要自己寫一個存儲系統,可以依照以下步驟: 1.寫一個繼承自django.core.files.storage.Storage的子類。 2.Django必須可以在無任何參數的情況下實例化MyStorage,所以任何環境設置必須來自django.conf.settings。 3.根據Storage的op ...
要自己寫一個存儲系統,可以依照以下步驟:
1.寫一個繼承自django.core.files.storage.Storage的子類。
from django.core.files.storage import Storage class MyStorage(Storage): ...
2.Django必須可以在無任何參數的情況下實例化MyStorage,所以任何環境設置必須來自django.conf.settings。
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 ...
3.根據Storage的open和save方法源碼:
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)
MyStorage需要實現_open和_save方法。
如果寫的是個本地存儲系統,還要重寫path方法。
4.使用django.utils.deconstruct.deconstructible裝飾器,以便在migration可以序列化。
還有,Storage.delete()、Storage.exists()、Storage.listdir()、Storage.size()、Storage.url()方法都會報NotImplementedError,也需要重寫。
Django Qiniu Storage
七牛雲有自己的django storage系統,可以看下是怎麼運作的,地址https://github.com/glasslion/django-qiniu-storage。
先在環境變數或者settings中配置QINIU_ACCESS_KEY、QINIU_SECRET_KEY、QINIU_BUCKET_NAME、QINIU_BUCKET_DOMAIN、QINIU_SECURE_URL。
使用七牛雲托管用戶上傳的文件,在 settings.py 里設置DEFAULT_FILE_STORAGE:
DEFAULT_FILE_STORAGE = 'qiniustorage.backends.QiniuStorage'
使用七牛托管動態生成的文件以及站點自身的靜態文件,設置:
STATICFILES_STORAGE = 'qiniustorage.backends.QiniuStaticStorage'
運行python manage.py collectstatic,靜態文件就會被統一上傳到七牛。
QiniuStorage代碼如下:
@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('/') def _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)
配置是從環境變數或者settings.py中獲得的:
def get_qiniu_config(name, default=None): """ Get configuration variable from environment variable or django setting.py """ config = os.environ.get(name, getattr(settings, name, default)) if config is not None: if isinstance(config, six.string_types): return config.strip() else: return config else: raise ImproperlyConfigured( "Can't find config for '%s' either in environment" "variable or in setting.py" % name) QINIU_ACCESS_KEY = get_qiniu_config('QINIU_ACCESS_KEY') QINIU_SECRET_KEY = get_qiniu_config('QINIU_SECRET_KEY') QINIU_BUCKET_NAME = get_qiniu_config('QINIU_BUCKET_NAME') QINIU_BUCKET_DOMAIN = get_qiniu_config('QINIU_BUCKET_DOMAIN', '').rstrip('/') QINIU_SECURE_URL = get_qiniu_config('QINIU_SECURE_URL', 'False')
重寫了_open和_save方法:
def _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
使用的put_data方法上傳文件,相關代碼如下:
def put_data( up_token, key, data, params=None, mime_type='application/octet-stream', check_crc=False, progress_handler=None, fname=None): """上傳二進位流到七牛 Args: up_token: 上傳憑證 key: 上傳文件名 data: 上傳二進位流 params: 自定義變數,規格參考 http://developer.qiniu.com/docs/v6/api/overview/up/response/vars.html#xvar mime_type: 上傳數據的mimeType check_crc: 是否校驗crc32 progress_handler: 上傳進度 Returns: 一個dict變數,類似 {"hash": "<Hash string>", "key": "<Key string>"} 一個ResponseInfo對象 """ crc = crc32(data) if check_crc else None return _form_put(up_token, key, data, params, mime_type, crc, progress_handler, fname) def _form_put(up_token, key, data, params, mime_type, crc, progress_handler=None, file_name=None): fields = {} if params: for k, v in params.items(): fields[k] = str(v) if crc: fields['crc32'] = crc if key is not None: fields['key'] = key fields['token'] = up_token url = config.get_default('default_zone').get_up_host_by_token(up_token) + '/' # name = key if key else file_name fname = file_name if not fname or not fname.strip(): fname = 'file_name' r, info = http._post_file(url, data=fields, files={'file': (fname, data, mime_type)}) if r is None and info.need_retry(): if info.connect_failed: url = config.get_default('default_zone').get_up_host_backup_by_token(up_token) + '/' if hasattr(data, 'read') is False: pass elif hasattr(data, 'seek') and (not hasattr(data, 'seekable') or data.seekable()): data.seek(0) else: return r, info r, info = http._post_file(url, data=fields, files={'file': (fname, data, mime_type)}) return r, info def _post_file(url, data, files): return _post(url, data, files, None) def _post(url, data, files, auth, headers=None): if _session is None: _init() try: post_headers = _headers.copy() if headers is not None: for k, v in headers.items(): post_headers.update({k: v}) r = _session.post( url, data=data, files=files, auth=auth, headers=post_headers, timeout=config.get_default('connection_timeout')) except Exception as e: return None, ResponseInfo(None, e) return __return_wrapper(r) def _init(): session = requests.Session() adapter = requests.adapters.HTTPAdapter( pool_connections=config.get_default('connection_pool'), pool_maxsize=config.get_default('connection_pool'), max_retries=config.get_default('connection_retries')) session.mount('http://', adapter) global _session _session = session
最終使用的是requests庫上傳文件的,統一適配了鏈接池個數、鏈接重試次數。