import os
import pathlib
from datetime import datetime, timezone
from urllib.parse import urljoin

from django.conf import settings
from django.core.exceptions import SuspiciousFileOperation
from django.core.files import File, locks
from django.core.files.move import file_move_safe
from django.core.files.utils import validate_file_name
from django.core.signals import setting_changed
from django.utils._os import safe_join
from django.utils.crypto import get_random_string
from django.utils.deconstruct import deconstructible
from django.utils.encoding import filepath_to_uri
from django.utils.functional import LazyObject, cached_property
from django.utils.module_loading import import_string
from django.utils.text import get_valid_filename

__all__ = (

[文档]class Storage: """ A base storage class, providing some default behaviors that all other storage systems can inherit or override, as necessary. """ # The following methods represent a public interface to private methods. # These shouldn't be overridden by subclasses unless absolutely necessary.
[文档] def open(self, name, mode="rb"): """Retrieve the specified file from storage.""" return self._open(name, mode)
[文档] def save(self, name, content, max_length=None): """ Save 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 = if not hasattr(content, "chunks"): content = File(content, name) name = self.get_available_name(name, max_length=max_length) name = self._save(name, content) # Ensure that the name returned from the storage system is still valid. validate_file_name(name, allow_relative_path=True) return name
# These methods are part of the public API, with default implementations.
[文档] def get_valid_name(self, name): """ Return a filename, based on the provided filename, that's suitable for use in the target storage system. """ return get_valid_filename(name)
[文档] def get_alternative_name(self, file_root, file_ext): """ Return an alternative filename, by adding an underscore and a random 7 character alphanumeric string (before the file extension, if one exists) to the filename. """ return "%s_%s%s" % (file_root, get_random_string(7), file_ext)
[文档] def get_available_name(self, name, max_length=None): """ Return a filename that's free on the target storage system and available for new content to be written to. """ name = str(name).replace("\\", "/") dir_name, file_name = os.path.split(name) if ".." in pathlib.PurePath(dir_name).parts: raise SuspiciousFileOperation( "Detected path traversal attempt in '%s'" % dir_name ) validate_file_name(file_name) file_root, file_ext = os.path.splitext(file_name) # If the filename already exists, generate an alternative filename # until it doesn't exist. # Truncate original name if required, so the new filename does not # exceed the max_length. while self.exists(name) or (max_length and len(name) > max_length): # file_ext includes the dot. name = os.path.join( dir_name, self.get_alternative_name(file_root, file_ext) ) if max_length is None: continue # Truncate file_root if max_length exceeded. truncation = len(name) - max_length if truncation > 0: file_root = file_root[:-truncation] # Entire file_root was truncated in attempt to find an # available filename. if not file_root: raise SuspiciousFileOperation( 'Storage can not find an available filename for "%s". ' "Please make sure that the corresponding file field " 'allows sufficient "max_length".' % name ) name = os.path.join( dir_name, self.get_alternative_name(file_root, file_ext) ) return name
[文档] def generate_filename(self, filename): """ Validate the filename by calling get_valid_name() and return a filename to be passed to the save() method. """ filename = str(filename).replace("\\", "/") # `filename` may include a path as returned by FileField.upload_to. dirname, filename = os.path.split(filename) if ".." in pathlib.PurePath(dirname).parts: raise SuspiciousFileOperation( "Detected path traversal attempt in '%s'" % dirname ) return os.path.normpath(os.path.join(dirname, self.get_valid_name(filename)))
[文档] def path(self, name): """ Return 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.")
# The following methods form the public API for storage systems, but with # no default implementations. Subclasses must implement *all* of these.
[文档] def delete(self, name): """ Delete the specified file from the storage system. """ raise NotImplementedError( "subclasses of Storage must provide a delete() method" )
[文档] def exists(self, name): """ Return True if a file referenced by the given name already exists in the storage system, or False if the name is available for a new file. """ raise NotImplementedError( "subclasses of Storage must provide an exists() method" )
[文档] def listdir(self, path): """ List the contents of the specified path. Return a 2-tuple of lists: the first item being directories, the second item being files. """ raise NotImplementedError( "subclasses of Storage must provide a listdir() method" )
[文档] def size(self, name): """ Return the total size, in bytes, of the file specified by name. """ raise NotImplementedError("subclasses of Storage must provide a size() method")
[文档] def url(self, name): """ Return an absolute URL where the file's contents can be accessed directly by a web browser. """ raise NotImplementedError("subclasses of Storage must provide a url() method")
[文档] def get_accessed_time(self, name): """ Return the last accessed time (as a datetime) of the file specified by name. The datetime will be timezone-aware if USE_TZ=True. """ raise NotImplementedError( "subclasses of Storage must provide a get_accessed_time() method" )
[文档] def get_created_time(self, name): """ Return the creation time (as a datetime) of the file specified by name. The datetime will be timezone-aware if USE_TZ=True. """ raise NotImplementedError( "subclasses of Storage must provide a get_created_time() method" )
[文档] def get_modified_time(self, name): """ Return the last modified time (as a datetime) of the file specified by name. The datetime will be timezone-aware if USE_TZ=True. """ raise NotImplementedError( "subclasses of Storage must provide a get_modified_time() method" )
[文档]@deconstructible class FileSystemStorage(Storage): """ Standard filesystem storage """ # The combination of O_CREAT and O_EXCL makes raise OSError if # the file already exists before it's opened. OS_OPEN_FLAGS = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0) def __init__( self, location=None, base_url=None, file_permissions_mode=None, directory_permissions_mode=None, ): self._location = location self._base_url = base_url self._file_permissions_mode = file_permissions_mode self._directory_permissions_mode = directory_permissions_mode setting_changed.connect(self._clear_cached_properties) def _clear_cached_properties(self, setting, **kwargs): """Reset setting based property values.""" if setting == "MEDIA_ROOT": self.__dict__.pop("base_location", None) self.__dict__.pop("location", None) elif setting == "MEDIA_URL": self.__dict__.pop("base_url", None) elif setting == "FILE_UPLOAD_PERMISSIONS": self.__dict__.pop("file_permissions_mode", None) elif setting == "FILE_UPLOAD_DIRECTORY_PERMISSIONS": self.__dict__.pop("directory_permissions_mode", None) def _value_or_setting(self, value, setting): return setting if value is None else value @cached_property def base_location(self): return self._value_or_setting(self._location, settings.MEDIA_ROOT) @cached_property def location(self): return os.path.abspath(self.base_location) @cached_property def base_url(self): if self._base_url is not None and not self._base_url.endswith("/"): self._base_url += "/" return self._value_or_setting(self._base_url, settings.MEDIA_URL) @cached_property def file_permissions_mode(self): return self._value_or_setting( self._file_permissions_mode, settings.FILE_UPLOAD_PERMISSIONS ) @cached_property def directory_permissions_mode(self): return self._value_or_setting( self._directory_permissions_mode, settings.FILE_UPLOAD_DIRECTORY_PERMISSIONS ) 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. directory = os.path.dirname(full_path) try: if self.directory_permissions_mode is not None: # Set the umask because os.makedirs() doesn't apply the "mode" # argument to intermediate-level directories. old_umask = os.umask(0o777 & ~self.directory_permissions_mode) try: os.makedirs( directory, self.directory_permissions_mode, exist_ok=True ) finally: os.umask(old_umask) else: os.makedirs(directory, exist_ok=True) except FileExistsError: raise FileExistsError("%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) # This is a normal uploadedfile that we can stream. else: # The current umask value is masked out by! fd =, self.OS_OPEN_FLAGS, 0o666) _file = None try: locks.lock(fd, locks.LOCK_EX) for chunk in content.chunks(): if _file is None: mode = "wb" if isinstance(chunk, bytes) else "wt" _file = os.fdopen(fd, mode) _file.write(chunk) finally: locks.unlock(fd) if _file is not None: _file.close() else: os.close(fd) except FileExistsError: # A new name is needed if the file exists. name = self.get_available_name(name) full_path = self.path(name) else: # OK, the file save worked. Break out of the loop. break if self.file_permissions_mode is not None: os.chmod(full_path, self.file_permissions_mode) # Ensure the saved path is always relative to the storage root. name = os.path.relpath(full_path, self.location) # Ensure the moved file has the same gid as the storage root. self._ensure_location_group_id(full_path) # Store filenames with forward slashes, even on Windows. return str(name).replace("\\", "/") def _ensure_location_group_id(self, full_path): if == "posix": file_gid = os.stat(full_path).st_gid location_gid = os.stat(self.location).st_gid if file_gid != location_gid: try: os.chown(full_path, uid=-1, gid=location_gid) except PermissionError: pass def delete(self, name): if not name: raise ValueError("The name must be given to delete().") name = self.path(name) # If the file or directory exists, delete it from the filesystem. try: if os.path.isdir(name): os.rmdir(name) else: os.remove(name) except FileNotFoundError: # FileNotFoundError is raised if the file or directory was removed # concurrently. pass def exists(self, name): return os.path.lexists(self.path(name)) def listdir(self, path): path = self.path(path) directories, files = [], [] with os.scandir(path) as entries: for entry in entries: if entry.is_dir(): directories.append( else: files.append( return directories, files def path(self, name): return safe_join(self.location, name) 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.") url = filepath_to_uri(name) if url is not None: url = url.lstrip("/") return urljoin(self.base_url, url) def _datetime_from_timestamp(self, ts): """ If timezone support is enabled, make an aware datetime object in UTC; otherwise make a naive one in the local timezone. """ tz = timezone.utc if settings.USE_TZ else None return datetime.fromtimestamp(ts, tz=tz) def get_accessed_time(self, name): return self._datetime_from_timestamp(os.path.getatime(self.path(name)))
[文档] def get_created_time(self, name): return self._datetime_from_timestamp(os.path.getctime(self.path(name)))
def get_modified_time(self, name): return self._datetime_from_timestamp(os.path.getmtime(self.path(name)))
[文档]def get_storage_class(import_path=None): return import_string(import_path or settings.DEFAULT_FILE_STORAGE)
[文档]class DefaultStorage(LazyObject): def _setup(self): self._wrapped = get_storage_class()()
default_storage = DefaultStorage()
Back to Top