Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import os
2import pathlib
4from django.core.exceptions import SuspiciousFileOperation
5from django.core.files import File
6from django.core.files.utils import validate_file_name
7from django.utils.crypto import get_random_string
8from django.utils.text import get_valid_filename
11class Storage:
12 """
13 A base storage class, providing some default behaviors that all other
14 storage systems can inherit or override, as necessary.
15 """
17 # The following methods represent a public interface to private methods.
18 # These shouldn't be overridden by subclasses unless absolutely necessary.
20 def open(self, name, mode="rb"):
21 """Retrieve the specified file from storage."""
22 return self._open(name, mode)
24 def save(self, name, content, max_length=None):
25 """
26 Save new content to the file specified by name. The content should be
27 a proper File object or any Python file-like object, ready to be read
28 from the beginning.
29 """
30 # Get the proper name for the file, as it will actually be saved.
31 if name is None:
32 name = content.name
34 if not hasattr(content, "chunks"):
35 content = File(content, name)
37 # Ensure that the name is valid, before and after having the storage
38 # system potentially modifying the name. This duplicates the check made
39 # inside `get_available_name` but it's necessary for those cases where
40 # `get_available_name` is overriden and validation is lost.
41 validate_file_name(name, allow_relative_path=True)
43 # Potentially find a different name depending on storage constraints.
44 name = self.get_available_name(name, max_length=max_length)
45 # Validate the (potentially) new name.
46 validate_file_name(name, allow_relative_path=True)
48 # The save operation should return the actual name of the file saved.
49 name = self._save(name, content)
50 # Ensure that the name returned from the storage system is still valid.
51 validate_file_name(name, allow_relative_path=True)
52 return name
54 def is_name_available(self, name, max_length=None):
55 exceeds_max_length = max_length and len(name) > max_length
56 return not self.exists(name) and not exceeds_max_length
58 # These methods are part of the public API, with default implementations.
60 def get_valid_name(self, name):
61 """
62 Return a filename, based on the provided filename, that's suitable for
63 use in the target storage system.
64 """
65 return get_valid_filename(name)
67 def get_alternative_name(self, file_root, file_ext):
68 """
69 Return an alternative filename, by adding an underscore and a random 7
70 character alphanumeric string (before the file extension, if one
71 exists) to the filename.
72 """
73 return "%s_%s%s" % (file_root, get_random_string(7), file_ext)
75 def get_available_name(self, name, max_length=None):
76 """
77 Return a filename that's free on the target storage system and
78 available for new content to be written to.
79 """
80 name = str(name).replace("\\", "/")
81 dir_name, file_name = os.path.split(name)
82 if ".." in pathlib.PurePath(dir_name).parts:
83 raise SuspiciousFileOperation(
84 "Detected path traversal attempt in '%s'" % dir_name
85 )
86 validate_file_name(file_name)
87 file_ext = "".join(pathlib.PurePath(file_name).suffixes)
88 file_root = file_name.removesuffix(file_ext)
89 # If the filename is not available, generate an alternative
90 # filename until one is available.
91 # Truncate original name if required, so the new filename does not
92 # exceed the max_length.
93 while not self.is_name_available(name, max_length=max_length):
94 # file_ext includes the dot.
95 name = os.path.join(
96 dir_name, self.get_alternative_name(file_root, file_ext)
97 )
98 if max_length is None:
99 continue
100 # Truncate file_root if max_length exceeded.
101 truncation = len(name) - max_length
102 if truncation > 0:
103 file_root = file_root[:-truncation]
104 # Entire file_root was truncated in attempt to find an
105 # available filename.
106 if not file_root:
107 raise SuspiciousFileOperation(
108 'Storage can not find an available filename for "%s". '
109 "Please make sure that the corresponding file field "
110 'allows sufficient "max_length".' % name
111 )
112 name = os.path.join(
113 dir_name, self.get_alternative_name(file_root, file_ext)
114 )
115 return name
117 def generate_filename(self, filename):
118 """
119 Validate the filename by calling get_valid_name() and return a filename
120 to be passed to the save() method.
121 """
122 filename = str(filename).replace("\\", "/")
123 # `filename` may include a path as returned by FileField.upload_to.
124 dirname, filename = os.path.split(filename)
125 if ".." in pathlib.PurePath(dirname).parts:
126 raise SuspiciousFileOperation(
127 "Detected path traversal attempt in '%s'" % dirname
128 )
129 return os.path.normpath(os.path.join(dirname, self.get_valid_name(filename)))
131 def path(self, name):
132 """
133 Return a local filesystem path where the file can be retrieved using
134 Python's built-in open() function. Storage systems that can't be
135 accessed using open() should *not* implement this method.
136 """
137 raise NotImplementedError("This backend doesn't support absolute paths.")
139 # The following methods form the public API for storage systems, but with
140 # no default implementations. Subclasses must implement *all* of these.
142 def delete(self, name):
143 """
144 Delete the specified file from the storage system.
145 """
146 raise NotImplementedError(
147 "subclasses of Storage must provide a delete() method"
148 )
150 def exists(self, name):
151 """
152 Return True if a file referenced by the given name already exists in the
153 storage system, or False if the name is available for a new file.
154 """
155 raise NotImplementedError(
156 "subclasses of Storage must provide an exists() method"
157 )
159 def listdir(self, path):
160 """
161 List the contents of the specified path. Return a 2-tuple of lists:
162 the first item being directories, the second item being files.
163 """
164 raise NotImplementedError(
165 "subclasses of Storage must provide a listdir() method"
166 )
168 def size(self, name):
169 """
170 Return the total size, in bytes, of the file specified by name.
171 """
172 raise NotImplementedError("subclasses of Storage must provide a size() method")
174 def url(self, name):
175 """
176 Return an absolute URL where the file's contents can be accessed
177 directly by a web browser.
178 """
179 raise NotImplementedError("subclasses of Storage must provide a url() method")
181 def get_accessed_time(self, name):
182 """
183 Return the last accessed time (as a datetime) of the file specified by
184 name. The datetime will be timezone-aware if USE_TZ=True.
185 """
186 raise NotImplementedError(
187 "subclasses of Storage must provide a get_accessed_time() method"
188 )
190 def get_created_time(self, name):
191 """
192 Return the creation time (as a datetime) of the file specified by name.
193 The datetime will be timezone-aware if USE_TZ=True.
194 """
195 raise NotImplementedError(
196 "subclasses of Storage must provide a get_created_time() method"
197 )
199 def get_modified_time(self, name):
200 """
201 Return the last modified time (as a datetime) of the file specified by
202 name. The datetime will be timezone-aware if USE_TZ=True.
203 """
204 raise NotImplementedError(
205 "subclasses of Storage must provide a get_modified_time() method"
206 )