首页 > 代码库 > django “如何”系列5：如何编写自定义存储系统
django “如何”系列5：如何编写自定义存储系统
如果你需要提供一个自定义的文件存储-一个常见的例子便是在远程系统上存储文件-你可以通过定义一个自己的存储类来做这件事情，你将通过一下步骤：
- 你自定义的存储系统一定是django.core.files.storage.Storage的子类
from django.core.files.storage import Storageclass MyStorage(Storage): ...
- django必须可以实例化你的存储系统（不使用任何参数），这意味着任何的设置都必须从django.conf.settings中拿
from django.conf import settingsfrom django.core.files.storage import Storageclass MyStorage(Storage): def __init__(self, option=None): if not option: option = settings.CUSTOM_STORAGE_OPTIONS
- 你的存储类必须实现_open()和_save()方法，以及其他的和你存储类相关的方法，下面会讲到
- 另外，如果你的类提供本地文件存储，你必须覆盖path()方法，如果不，可以忽略这个方法
这是基类Storage的源码，我会在源码的注释中讲解一些内容一些要注意的内容
class Storage(object): """ 存储基类，提供一些默认的行为供其他的存储系统继承或者覆盖（如果需要的话） """ # 下面的方法代表了私有方法的一个公共接口，除非绝对的需要，这些方法不应该被子类覆盖 def open(self, name, mode=‘rb‘): """ 从存储中检索特定的文件 """ return self._open(name, mode) def save(self, name, content): """ 用给定的文件名保存给定的新内容，内容应该是一个合适的可以从头开始读取的File对象 """ # Get the proper name for the file, as it will actually be saved. if name is None: name = content.name name = self.get_available_name(name) name = self._save(name, content) # Store filenames with forward slashes, even on Windows return force_unicode(name.replace(‘\\‘, ‘/‘)) # 这些方法是一部分的公共API（已经实现好的，当然，你可以覆盖） def get_valid_name(self, name): """ Returns a filename, based on the provided filename, that‘s suitable for use in the target storage system. """ return get_valid_filename(name) def get_available_name(self, name): """ Returns a filename that‘s free on the target storage system, and available for new content to be written to. """ dir_name, file_name = os.path.split(name) file_root, file_ext = os.path.splitext(file_name) # If the filename already exists, add an underscore and a number (before # the file extension, if one exists) to the filename until the generated # filename doesn‘t exist. count = itertools.count(1) while self.exists(name): # file_ext includes the dot. name = os.path.join(dir_name, "%s_%s%s" % (file_root, count.next(), file_ext)) return name def path(self, name): """ Returns a local filesystem path where the file can be retrieved using Python‘s built-in open() function. Storage systems that can‘t be accessed using open() should *not* implement this method. """ raise NotImplementedError("This backend doesn‘t support absolute paths.") # 下面的这些方法是没有提供默认实现的公共API，子类一定要实现这些方法 def delete(self, name): """ Deletes the specified file from the storage system. """ raise NotImplementedError() def exists(self, name): """ Returns True if a file referened by the given name already exists in the storage system, or False if the name is available for a new file. """ raise NotImplementedError() def listdir(self, path): """ Lists the contents of the specified path, returning a 2-tuple of lists; the first item being directories, the second item being files. """ raise NotImplementedError() def size(self, name): """ Returns the total size, in bytes, of the file specified by name. """ raise NotImplementedError() def url(self, name): """ Returns an absolute URL where the file‘s contents can be accessed directly by a Web browser. """ raise NotImplementedError() def accessed_time(self, name): """ Returns the last accessed time (as datetime object) of the file specified by name. """ raise NotImplementedError() def created_time(self, name): """ Returns the creation time (as datetime object) of the file specified by name. """ raise NotImplementedError() def modified_time(self, name): """ Returns the last modified time (as datetime object) of the file specified by name. """ raise NotImplementedError()
看完这个源码，相信你已经知道该如何如写一个自己的存储系统类了（那些方法一定要有的，那些是可以直接用的，那些是一定要写的），下面我们看一下django自带的一个实现吧
class FileSystemStorage(Storage): """ Standard filesystem storage """ def __init__(self, location=None, base_url=None): if location is None: location = settings.MEDIA_ROOT self.base_location = location self.location = abspathu(self.base_location) if base_url is None: base_url = settings.MEDIA_URL self.base_url = base_url def _open(self, name, mode=‘rb‘): return File(open(self.path(name), mode)) def _save(self, name, content): full_path = self.path(name) # Create any intermediate directories that do not exist. # Note that there is a race between os.path.exists and os.makedirs: # if os.makedirs fails with EEXIST, the directory was created # concurrently, and we can continue normally. Refs #16082. directory = os.path.dirname(full_path) if not os.path.exists(directory): try: os.makedirs(directory) except OSError, e: if e.errno != errno.EEXIST: raise if not os.path.isdir(directory): raise IOError("%s exists and is not a directory." % directory) # There‘s a potential race condition between get_available_name and # saving the file; it‘s possible that two threads might return the # same name, at which point all sorts of fun happens. So we need to # try to create the file, but if it already exists we have to go back # to get_available_name() and try again. while True: try: # This file has a file path that we can move. if hasattr(content, ‘temporary_file_path‘): file_move_safe(content.temporary_file_path(), full_path) content.close() # This is a normal uploadedfile that we can stream. else: # This fun binary flag incantation makes os.open throw an # OSError if the file already exists before we open it. fd = os.open(full_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, ‘O_BINARY‘, 0)) try: locks.lock(fd, locks.LOCK_EX) for chunk in content.chunks(): os.write(fd, chunk) finally: locks.unlock(fd) os.close(fd) except OSError, e: if e.errno == errno.EEXIST: # Ooops, the file exists. We need a new file name. name = self.get_available_name(name) full_path = self.path(name) else: raise else: # OK, the file save worked. Break out of the loop. break if settings.FILE_UPLOAD_PERMISSIONS is not None: os.chmod(full_path, settings.FILE_UPLOAD_PERMISSIONS) return name def delete(self, name): name = self.path(name) # If the file exists, delete it from the filesystem. # Note that there is a race between os.path.exists and os.remove: # if os.remove fails with ENOENT, the file was removed # concurrently, and we can continue normally. if os.path.exists(name): try: os.remove(name) except OSError, e: if e.errno != errno.ENOENT: raise def exists(self, name): return os.path.exists(self.path(name)) def listdir(self, path): path = self.path(path) directories, files = [], [] for entry in os.listdir(path): if os.path.isdir(os.path.join(path, entry)): directories.append(entry) else: files.append(entry) return directories, files def path(self, name): try: path = safe_join(self.location, name) except ValueError: raise SuspiciousOperation("Attempted access to ‘%s‘ denied." % name) return os.path.normpath(path) def size(self, name): return os.path.getsize(self.path(name)) def url(self, name): if self.base_url is None: raise ValueError("This file is not accessible via a URL.") return urlparse.urljoin(self.base_url, filepath_to_uri(name)) def accessed_time(self, name): return datetime.fromtimestamp(os.path.getatime(self.path(name))) def created_time(self, name): return datetime.fromtimestamp(os.path.getctime(self.path(name))) def modified_time(self, name): return datetime.fromtimestamp(os.path.getmtime(self.path(name)))
django “如何”系列5：如何编写自定义存储系统
声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉： 投诉/举报 工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。