Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/django/core/files/uploadhandler.py: 76%
78 statements
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
1"""
2Base file upload handler classes, and the built-in concrete subclasses
3"""
4import os
5from io import BytesIO
7from django.conf import settings
8from django.core.files.uploadedfile import InMemoryUploadedFile, TemporaryUploadedFile
9from django.utils.module_loading import import_string
11__all__ = [
12 "UploadFileException",
13 "StopUpload",
14 "SkipFile",
15 "FileUploadHandler",
16 "TemporaryFileUploadHandler",
17 "MemoryFileUploadHandler",
18 "load_handler",
19 "StopFutureHandlers",
20]
23class UploadFileException(Exception):
24 """
25 Any error having to do with uploading files.
26 """
28 pass
31class StopUpload(UploadFileException):
32 """
33 This exception is raised when an upload must abort.
34 """
36 def __init__(self, connection_reset=False):
37 """
38 If ``connection_reset`` is ``True``, Django knows will halt the upload
39 without consuming the rest of the upload. This will cause the browser to
40 show a "connection reset" error.
41 """
42 self.connection_reset = connection_reset
44 def __str__(self):
45 if self.connection_reset:
46 return "StopUpload: Halt current upload."
47 else:
48 return "StopUpload: Consume request data, then halt."
51class SkipFile(UploadFileException):
52 """
53 This exception is raised by an upload handler that wants to skip a given file.
54 """
56 pass
59class StopFutureHandlers(UploadFileException):
60 """
61 Upload handlers that have handled a file and do not want future handlers to
62 run should raise this exception instead of returning None.
63 """
65 pass
68class FileUploadHandler:
69 """
70 Base class for streaming upload handlers.
71 """
73 chunk_size = 64 * 2**10 # : The default chunk size is 64 KB.
75 def __init__(self, request=None):
76 self.file_name = None
77 self.content_type = None
78 self.content_length = None
79 self.charset = None
80 self.content_type_extra = None
81 self.request = request
83 def handle_raw_input(
84 self, input_data, META, content_length, boundary, encoding=None
85 ):
86 """
87 Handle the raw input from the client.
89 Parameters:
91 :input_data:
92 An object that supports reading via .read().
93 :META:
94 ``request.META``.
95 :content_length:
96 The (integer) value of the Content-Length header from the
97 client.
98 :boundary: The boundary from the Content-Type header. Be sure to
99 prepend two '--'.
100 """
101 pass
103 def new_file(
104 self,
105 field_name,
106 file_name,
107 content_type,
108 content_length,
109 charset=None,
110 content_type_extra=None,
111 ):
112 """
113 Signal that a new file has been started.
115 Warning: As with any data from the client, you should not trust
116 content_length (and sometimes won't even get it).
117 """
118 self.field_name = field_name
119 self.file_name = file_name
120 self.content_type = content_type
121 self.content_length = content_length
122 self.charset = charset
123 self.content_type_extra = content_type_extra
125 def receive_data_chunk(self, raw_data, start):
126 """
127 Receive data from the streamed upload parser. ``start`` is the position
128 in the file of the chunk.
129 """
130 raise NotImplementedError(
131 "subclasses of FileUploadHandler must provide a receive_data_chunk() method"
132 )
134 def file_complete(self, file_size):
135 """
136 Signal that a file has completed. File size corresponds to the actual
137 size accumulated by all the chunks.
139 Subclasses should return a valid ``UploadedFile`` object.
140 """
141 raise NotImplementedError(
142 "subclasses of FileUploadHandler must provide a file_complete() method"
143 )
145 def upload_complete(self):
146 """
147 Signal that the upload is complete. Subclasses should perform cleanup
148 that is necessary for this handler.
149 """
150 pass
152 def upload_interrupted(self):
153 """
154 Signal that the upload was interrupted. Subclasses should perform
155 cleanup that is necessary for this handler.
156 """
157 pass
160class TemporaryFileUploadHandler(FileUploadHandler):
161 """
162 Upload handler that streams data into a temporary file.
163 """
165 def new_file(self, *args, **kwargs):
166 """
167 Create the file object to append to as data is coming in.
168 """
169 super().new_file(*args, **kwargs)
170 self.file = TemporaryUploadedFile(
171 self.file_name, self.content_type, 0, self.charset, self.content_type_extra
172 )
174 def receive_data_chunk(self, raw_data, start):
175 self.file.write(raw_data)
177 def file_complete(self, file_size):
178 self.file.seek(0)
179 self.file.size = file_size
180 return self.file
182 def upload_interrupted(self):
183 if hasattr(self, "file"):
184 temp_location = self.file.temporary_file_path()
185 try:
186 self.file.close()
187 os.remove(temp_location)
188 except FileNotFoundError:
189 pass
192class MemoryFileUploadHandler(FileUploadHandler):
193 """
194 File upload handler to stream uploads into memory (used for small files).
195 """
197 def handle_raw_input(
198 self, input_data, META, content_length, boundary, encoding=None
199 ):
200 """
201 Use the content_length to signal whether or not this handler should be
202 used.
203 """
204 # Check the content-length header to see if we should
205 # If the post is too large, we cannot use the Memory handler.
206 self.activated = content_length <= settings.FILE_UPLOAD_MAX_MEMORY_SIZE
208 def new_file(self, *args, **kwargs):
209 super().new_file(*args, **kwargs)
210 if self.activated: 210 ↛ exitline 210 didn't return from function 'new_file', because the condition on line 210 was never false
211 self.file = BytesIO()
212 raise StopFutureHandlers()
214 def receive_data_chunk(self, raw_data, start):
215 """Add the data to the BytesIO file."""
216 if self.activated: 216 ↛ 219line 216 didn't jump to line 219, because the condition on line 216 was never false
217 self.file.write(raw_data)
218 else:
219 return raw_data
221 def file_complete(self, file_size):
222 """Return a file object if this handler is activated."""
223 if not self.activated: 223 ↛ 224line 223 didn't jump to line 224, because the condition on line 223 was never true
224 return
226 self.file.seek(0)
227 return InMemoryUploadedFile(
228 file=self.file,
229 field_name=self.field_name,
230 name=self.file_name,
231 content_type=self.content_type,
232 size=file_size,
233 charset=self.charset,
234 content_type_extra=self.content_type_extra,
235 )
238def load_handler(path, *args, **kwargs):
239 """
240 Given a path to a handler, return an instance of that handler.
242 E.g.::
243 >>> from django.http import HttpRequest
244 >>> request = HttpRequest()
245 >>> load_handler(
246 ... 'django.core.files.uploadhandler.TemporaryFileUploadHandler',
247 ... request,
248 ... )
249 <TemporaryFileUploadHandler object at 0x...>
250 """
251 return import_string(path)(*args, **kwargs)