# Django 3.2.12 documentation

# Source code for django.core.files.storage

    import os
    import pathlib
    from datetime import datetime
    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 import timezone
    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__ = (
        'Storage', 'FileSystemStorage', 'DefaultStorage', 'default_storage',
        'get_storage_class',
    )

    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 = content.name

            if not hasattr(content, 'chunks'):
                content = File(content, name)

            # Ensure that the name is valid, before and after having the storage
            # system potentially modifying the name. This duplicates the check made
            # inside `get_available_name` but it's necessary for those cases where
            # `get_available_name` is overriden and validation is lost.
            validate_file_name(name, allow_relative_path=True)

            # Potentially find a different name depending on storage constraints.
            name = self.get_available_name(name, max_length=max_length)
            # Validate the (potentially) new name.
            validate_file_name(name, allow_relative_path=True)

            # The save operation should return the actual name of the file saved.
            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 os.open() 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 os.open!
                        fd = os.open(full_path, 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)
            # Store filenames with forward slashes, even on Windows.
            return str(name).replace('\\', '/')

        def delete(self, name):
            assert name, "The name argument is not allowed to be empty."
            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.exists(self.path(name))

        def listdir(self, path):
            path = self.path(path)
            directories, files = [], []
            for entry in os.scandir(path):
                if entry.is_dir():
                    directories.append(entry.name)
                else:
                    files.append(entry.name)
            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.
            """
            if settings.USE_TZ:
                # Safe to use .replace() because UTC doesn't have DST
                return datetime.utcfromtimestamp(ts).replace(tzinfo=timezone.utc)
            else:
                return datetime.fromtimestamp(ts)

        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()