Fixed #35326 -- Added allow_overwrite parameter to FileSystemStorage.

django/core/files/storage/filesystem.py

@@ -1,13 +1,16 @@
 import os
+import warnings
 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.signals import setting_changed
 from django.utils._os import safe_join
 from django.utils.deconstruct import deconstructible
+from django.utils.deprecation import RemovedInDjango60Warning
 from django.utils.encoding import filepath_to_uri
 from django.utils.functional import cached_property
 
@@ -21,8 +24,7 @@ class FileSystemStorage(Storage, StorageSettingsMixin):
     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.
+    # RemovedInDjango60Warning: remove OS_OPEN_FLAGS.
     OS_OPEN_FLAGS = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0)
 
     def __init__(
@@ -31,12 +33,23 @@ def __init__(
         base_url=None,
         file_permissions_mode=None,
         directory_permissions_mode=None,
+        allow_overwrite=False,
     ):
         self._location = location
         self._base_url = base_url
         self._file_permissions_mode = file_permissions_mode
         self._directory_permissions_mode = directory_permissions_mode
+        self._allow_overwrite = allow_overwrite
         setting_changed.connect(self._clear_cached_properties)
+        # RemovedInDjango60Warning: remove this warning.
+        if self.OS_OPEN_FLAGS != os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(
+            os, "O_BINARY", 0
+        ):
+            warnings.warn(
+                "Overriding OS_OPEN_FLAGS is deprecated. Use "
+                "the allow_overwrite parameter instead.",
+                RemovedInDjango60Warning,
+            )
 
     @cached_property
     def base_location(self):
@@ -98,12 +111,30 @@ def _save(self, name, content):
             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)
+                    file_move_safe(
+                        content.temporary_file_path(),
+                        full_path,
+                        allow_overwrite=self._allow_overwrite,
+                    )
 
                 # 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)
+                    # The combination of O_CREAT and O_EXCL makes os.open() raises an
+                    # OSError if the file already exists before it's opened.
+                    open_flags = (
+                        os.O_WRONLY
+                        | os.O_CREAT
+                        | os.O_EXCL
+                        | getattr(os, "O_BINARY", 0)
+                    )
+                    # RemovedInDjango60Warning: when the deprecation ends, replace with:
+                    # if self._allow_overwrite:
+                    #     open_flags = open_flags & ~os.O_EXCL
+                    if self.OS_OPEN_FLAGS != open_flags:
+                        open_flags = self.OS_OPEN_FLAGS
+                    elif self._allow_overwrite:
+                        open_flags = open_flags & ~os.O_EXCL
+                    fd = os.open(full_path, open_flags, 0o666)
                     _file = None
                     try:
                         locks.lock(fd, locks.LOCK_EX)
@@ -162,7 +193,13 @@ def delete(self, name):
             pass
 
     def exists(self, name):
-        return os.path.lexists(self.path(name))
+        try:
+            exists = os.path.lexists(self.path(name))
+        except SuspiciousFileOperation:
+            raise
+        if self._allow_overwrite:
+            return False
+        return exists
 
     def listdir(self, path):
         path = self.path(path)

docs/internals/deprecation.txt

@@ -79,6 +79,9 @@ details on these changes.
 
 * The ``check`` keyword argument of ``CheckConstraint`` will be removed.
 
+* The ``OS_OPEN_FLAGS`` attribute of
+  :class:`~django.core.files.storage.FileSystemStorage` will be removed.
+
 .. _deprecation-removed-in-5.1:
 
 5.1

docs/ref/files/storage.txt

@@ -28,7 +28,7 @@ Django provides convenient ways to access the default storage class:
 The ``FileSystemStorage`` class
 ===============================
 
-.. class:: FileSystemStorage(location=None, base_url=None, file_permissions_mode=None, directory_permissions_mode=None)
+.. class:: FileSystemStorage(location=None, base_url=None, file_permissions_mode=None, directory_permissions_mode=None, allow_overwrite=False)
 
     The :class:`~django.core.files.storage.FileSystemStorage` class implements
     basic file storage on a local filesystem. It inherits from
@@ -60,6 +60,13 @@ The ``FileSystemStorage`` class
         The file system permissions that the directory will receive when it is
         saved. Defaults to :setting:`FILE_UPLOAD_DIRECTORY_PERMISSIONS`.
 
+    .. attribute:: allow_overwrite
+
+        .. versionadded:: 5.1
+
+        Flag to control allowing saving a new file over an existing one.
+        Defaults to ``False``.
+
     .. method:: get_created_time(name)
 
         Returns a :class:`~datetime.datetime` of the system's ctime, i.e.

docs/releases/5.1.txt

@@ -210,7 +210,10 @@ Error Reporting
 File Storage
 ~~~~~~~~~~~~
 
-* ...
+* The :attr:`~django.core.files.storage.FileSystemStorage.allow_overwrite`
+  parameter has been added to
+  :class:`~django.core.files.storage.FileSystemStorage`, to allow saving new
+  files over existing ones.
 
 File Uploads
 ~~~~~~~~~~~~
@@ -467,6 +470,12 @@ Miscellaneous
 * The ``check`` keyword argument of ``CheckConstraint`` is deprecated in favor
   of ``condition``.
 
+* The undocumented ``OS_OPEN_FLAGS`` property of
+  :class:`~django.core.files.storage.FileSystemStorage` has been deprecated.
+  To allow overwriting files in storage, set the new
+  :attr:`~django.core.files.storage.FileSystemStorage.allow_overwrite` option
+  to ``True`` instead.
+
 Features removed in 5.1
 =======================
 

tests/file_storage/tests.py

@@ -25,11 +25,18 @@
 )
 from django.db.models import FileField
 from django.db.models.fields.files import FileDescriptor
-from django.test import LiveServerTestCase, SimpleTestCase, TestCase, override_settings
+from django.test import (
+    LiveServerTestCase,
+    SimpleTestCase,
+    TestCase,
+    ignore_warnings,
+    override_settings,
+)
 from django.test.utils import requires_tz_support
 from django.urls import NoReverseMatch, reverse_lazy
 from django.utils import timezone
 from django.utils._os import symlinks_supported
+from django.utils.deprecation import RemovedInDjango60Warning
 
 from .models import (
     Storage,
@@ -88,18 +95,18 @@ def test_file_access_options(self):
         """
         Standard file access options are available, and work as expected.
         """
-        self.assertFalse(self.storage.exists("storage_test"))
+        self.assertFalse(os.path.exists(os.path.join(self.temp_dir, "storage_test")))
         f = self.storage.open("storage_test", "w")
         f.write("storage contents")
         f.close()
-        self.assertTrue(self.storage.exists("storage_test"))
+        self.assertTrue(os.path.exists(os.path.join(self.temp_dir, "storage_test")))
 
         f = self.storage.open("storage_test", "r")
         self.assertEqual(f.read(), "storage contents")
         f.close()
 
         self.storage.delete("storage_test")
-        self.assertFalse(self.storage.exists("storage_test"))
+        self.assertFalse(os.path.exists(os.path.join(self.temp_dir, "storage_test")))
 
     def _test_file_time_getter(self, getter):
         # Check for correct behavior under both USE_TZ=True and USE_TZ=False.
@@ -268,10 +275,10 @@ def test_file_save_with_path(self):
         """
         Saving a pathname should create intermediate directories as necessary.
         """
-        self.assertFalse(self.storage.exists("path/to"))
+        self.assertFalse(os.path.exists(os.path.join(self.temp_dir, "path/to")))
         self.storage.save("path/to/test.file", ContentFile("file saved with path"))
 
-        self.assertTrue(self.storage.exists("path/to"))
+        self.assertTrue(os.path.exists(os.path.join(self.temp_dir, "path/to")))
         with self.storage.open("path/to/test.file") as f:
             self.assertEqual(f.read(), b"file saved with path")
 
@@ -607,6 +614,7 @@ def test_custom_get_available_name(self):
         self.storage.delete(second)
 
 
+# RemovedInDjango60Warning: Remove this class.
 class OverwritingStorage(FileSystemStorage):
     """
     Overwrite existing files instead of appending a suffix to generate an
@@ -621,7 +629,26 @@ def get_available_name(self, name, max_length=None):
         return name
 
 
-class OverwritingStorageTests(FileStorageTests):
+# RemovedInDjango60Warning: Remove this test class.
+class OverwritingStorageOSOpenFlagsWarningTests(SimpleTestCase):
+    storage_class = OverwritingStorage
+
+    def setUp(self):
+        self.temp_dir = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.temp_dir)
+
+    def test_os_open_flags_deprecation_warning(self):
+        msg = "Overriding OS_OPEN_FLAGS is deprecated. Use the allow_overwrite "
+        msg += "parameter instead."
+        with self.assertWarnsMessage(RemovedInDjango60Warning, msg):
+            self.storage = self.storage_class(
+                location=self.temp_dir, base_url="/test_media_url/"
+            )
+
+
+# RemovedInDjango60Warning: Remove this test class.
+@ignore_warnings(category=RemovedInDjango60Warning)
+class OverwritingStorageOSOpenFlagsTests(FileStorageTests):
     storage_class = OverwritingStorage
 
     def test_save_overwrite_behavior(self):
@@ -649,6 +676,65 @@ def test_save_overwrite_behavior(self):
             self.storage.delete(name)
 
 
+class OverwritingStorageTests(FileStorageTests):
+    storage_class = FileSystemStorage
+
+    def setUp(self):
+        self.temp_dir = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.temp_dir)
+        self.storage = self.storage_class(
+            location=self.temp_dir, base_url="/test_media_url/", allow_overwrite=True
+        )
+
+    def test_save_overwrite_behavior(self):
+        """Saving to same file name twice overwrites the first file."""
+        name = "test.file"
+        self.assertFalse(self.storage.exists(name))
+        content_1 = b"content one"
+        content_2 = b"second content"
+        f_1 = ContentFile(content_1)
+        f_2 = ContentFile(content_2)
+        stored_name_1 = self.storage.save(name, f_1)
+        try:
+            self.assertEqual(stored_name_1, name)
+            self.assertTrue(os.path.exists(os.path.join(self.temp_dir, name)))
+            with self.storage.open(name) as fp:
+                self.assertEqual(fp.read(), content_1)
+            stored_name_2 = self.storage.save(name, f_2)
+            self.assertEqual(stored_name_2, name)
+            self.assertTrue(os.path.exists(os.path.join(self.temp_dir, name)))
+            with self.storage.open(name) as fp:
+                self.assertEqual(fp.read(), content_2)
+        finally:
+            self.storage.delete(name)
+
+    def test_save_overwrite_behavior_temp_file(self):
+        """Saving to same file name twice overwrites the first file."""
+        name = "test.file"
+        self.assertFalse(self.storage.exists(name))
+        content_1 = b"content one"
+        content_2 = b"second content"
+        f_1 = TemporaryUploadedFile("tmp1", "text/plain", 11, "utf8")
+        f_1.write(content_1)
+        f_1.seek(0)
+        f_2 = TemporaryUploadedFile("tmp2", "text/plain", 14, "utf8")
+        f_2.write(content_2)
+        f_2.seek(0)
+        stored_name_1 = self.storage.save(name, f_1)
+        try:
+            self.assertEqual(stored_name_1, name)
+            self.assertTrue(os.path.exists(os.path.join(self.temp_dir, name)))
+            with self.storage.open(name) as fp:
+                self.assertEqual(fp.read(), content_1)
+            stored_name_2 = self.storage.save(name, f_2)
+            self.assertEqual(stored_name_2, name)
+            self.assertTrue(os.path.exists(os.path.join(self.temp_dir, name)))
+            with self.storage.open(name) as fp:
+                self.assertEqual(fp.read(), content_2)
+        finally:
+            self.storage.delete(name)
+
+
 class DiscardingFalseContentStorage(FileSystemStorage):
     def _save(self, name, content):
         if content: