Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.10/site-packages/django/http/response.py: 37%
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 datetime
2import io
3import json
4import mimetypes
5import os
6import re
7import sys
8import time
9import warnings
10from email.header import Header
11from http.client import responses
12from urllib.parse import urlsplit
14from asgiref.sync import async_to_sync, sync_to_async
16from django.conf import settings
17from django.core import signals, signing
18from django.core.exceptions import DisallowedRedirect
19from django.core.serializers.json import DjangoJSONEncoder
20from django.http.cookie import SimpleCookie
21from django.utils import timezone
22from django.utils.datastructures import CaseInsensitiveMapping
23from django.utils.encoding import iri_to_uri
24from django.utils.functional import cached_property
25from django.utils.http import content_disposition_header, http_date
26from django.utils.regex_helper import _lazy_re_compile
28_charset_from_content_type_re = _lazy_re_compile(
29 r";\s*charset=(?P<charset>[^\s;]+)", re.I
30)
33class ResponseHeaders(CaseInsensitiveMapping):
34 def __init__(self, data):
35 """
36 Populate the initial data using __setitem__ to ensure values are
37 correctly encoded.
38 """
39 self._store = {}
40 if data:
41 for header, value in self._unpack_items(data):
42 self[header] = value
44 def _convert_to_charset(self, value, charset, mime_encode=False):
45 """
46 Convert headers key/value to ascii/latin-1 native strings.
47 `charset` must be 'ascii' or 'latin-1'. If `mime_encode` is True and
48 `value` can't be represented in the given charset, apply MIME-encoding.
49 """
50 try:
51 if isinstance(value, str):
52 # Ensure string is valid in given charset
53 value.encode(charset)
54 elif isinstance(value, bytes):
55 # Convert bytestring using given charset
56 value = value.decode(charset)
57 else:
58 value = str(value)
59 # Ensure string is valid in given charset.
60 value.encode(charset)
61 if "\n" in value or "\r" in value:
62 raise BadHeaderError(
63 f"Header values can't contain newlines (got {value!r})"
64 )
65 except UnicodeError as e:
66 # Encoding to a string of the specified charset failed, but we
67 # don't know what type that value was, or if it contains newlines,
68 # which we may need to check for before sending it to be
69 # encoded for multiple character sets.
70 if (isinstance(value, bytes) and (b"\n" in value or b"\r" in value)) or (
71 isinstance(value, str) and ("\n" in value or "\r" in value)
72 ):
73 raise BadHeaderError(
74 f"Header values can't contain newlines (got {value!r})"
75 ) from e
76 if mime_encode:
77 value = Header(value, "utf-8", maxlinelen=sys.maxsize).encode()
78 else:
79 e.reason += ", HTTP response headers must be in %s format" % charset
80 raise
81 return value
83 def __delitem__(self, key):
84 self.pop(key)
86 def __setitem__(self, key, value):
87 key = self._convert_to_charset(key, "ascii")
88 value = self._convert_to_charset(value, "latin-1", mime_encode=True)
89 self._store[key.lower()] = (key, value)
91 def pop(self, key, default=None):
92 return self._store.pop(key.lower(), default)
94 def setdefault(self, key, value):
95 if key not in self:
96 self[key] = value
99class BadHeaderError(ValueError):
100 pass
103class HttpResponseBase:
104 """
105 An HTTP response base class with dictionary-accessed headers.
107 This class doesn't handle content. It should not be used directly.
108 Use the HttpResponse and StreamingHttpResponse subclasses instead.
109 """
111 status_code = 200
113 def __init__(
114 self, content_type=None, status=None, reason=None, charset=None, headers=None
115 ):
116 self.headers = ResponseHeaders(headers)
117 self._charset = charset
118 if "Content-Type" not in self.headers:
119 if content_type is None:
120 content_type = f"text/html; charset={self.charset}"
121 self.headers["Content-Type"] = content_type
122 elif content_type:
123 raise ValueError(
124 "'headers' must not contain 'Content-Type' when the "
125 "'content_type' parameter is provided."
126 )
127 self._resource_closers = []
128 # This parameter is set by the handler. It's necessary to preserve the
129 # historical behavior of request_finished.
130 self._handler_class = None
131 self.cookies = SimpleCookie()
132 self.closed = False
133 if status is not None:
134 try:
135 self.status_code = int(status)
136 except (ValueError, TypeError):
137 raise TypeError("HTTP status code must be an integer.")
139 if not 100 <= self.status_code <= 599:
140 raise ValueError("HTTP status code must be an integer from 100 to 599.")
141 self._reason_phrase = reason
143 @property
144 def reason_phrase(self):
145 if self._reason_phrase is not None:
146 return self._reason_phrase
147 # Leave self._reason_phrase unset in order to use the default
148 # reason phrase for status code.
149 return responses.get(self.status_code, "Unknown Status Code")
151 @reason_phrase.setter
152 def reason_phrase(self, value):
153 self._reason_phrase = value
155 @property
156 def charset(self):
157 if self._charset is not None:
158 return self._charset
159 # The Content-Type header may not yet be set, because the charset is
160 # being inserted *into* it.
161 if content_type := self.headers.get("Content-Type"):
162 if matched := _charset_from_content_type_re.search(content_type):
163 # Extract the charset and strip its double quotes.
164 # Note that having parsed it from the Content-Type, we don't
165 # store it back into the _charset for later intentionally, to
166 # allow for the Content-Type to be switched again later.
167 return matched["charset"].replace('"', "")
168 return settings.DEFAULT_CHARSET
170 @charset.setter
171 def charset(self, value):
172 self._charset = value
174 def serialize_headers(self):
175 """HTTP headers as a bytestring."""
176 return b"\r\n".join(
177 [
178 key.encode("ascii") + b": " + value.encode("latin-1")
179 for key, value in self.headers.items()
180 ]
181 )
183 __bytes__ = serialize_headers
185 @property
186 def _content_type_for_repr(self):
187 return (
188 ', "%s"' % self.headers["Content-Type"]
189 if "Content-Type" in self.headers
190 else ""
191 )
193 def __setitem__(self, header, value):
194 self.headers[header] = value
196 def __delitem__(self, header):
197 del self.headers[header]
199 def __getitem__(self, header):
200 return self.headers[header]
202 def has_header(self, header):
203 """Case-insensitive check for a header."""
204 return header in self.headers
206 __contains__ = has_header
208 def items(self):
209 return self.headers.items()
211 def get(self, header, alternate=None):
212 return self.headers.get(header, alternate)
214 def set_cookie(
215 self,
216 key,
217 value="",
218 max_age=None,
219 expires=None,
220 path="/",
221 domain=None,
222 secure=False,
223 httponly=False,
224 samesite=None,
225 ):
226 """
227 Set a cookie.
229 ``expires`` can be:
230 - a string in the correct format,
231 - a naive ``datetime.datetime`` object in UTC,
232 - an aware ``datetime.datetime`` object in any time zone.
233 If it is a ``datetime.datetime`` object then calculate ``max_age``.
235 ``max_age`` can be:
236 - int/float specifying seconds,
237 - ``datetime.timedelta`` object.
238 """
239 self.cookies[key] = value
240 if expires is not None:
241 if isinstance(expires, datetime.datetime):
242 if timezone.is_naive(expires):
243 expires = timezone.make_aware(expires, datetime.timezone.utc)
244 delta = expires - datetime.datetime.now(tz=datetime.timezone.utc)
245 # Add one second so the date matches exactly (a fraction of
246 # time gets lost between converting to a timedelta and
247 # then the date string).
248 delta += datetime.timedelta(seconds=1)
249 # Just set max_age - the max_age logic will set expires.
250 expires = None
251 if max_age is not None:
252 raise ValueError("'expires' and 'max_age' can't be used together.")
253 max_age = max(0, delta.days * 86400 + delta.seconds)
254 else:
255 self.cookies[key]["expires"] = expires
256 else:
257 self.cookies[key]["expires"] = ""
258 if max_age is not None:
259 if isinstance(max_age, datetime.timedelta):
260 max_age = max_age.total_seconds()
261 self.cookies[key]["max-age"] = int(max_age)
262 # IE requires expires, so set it if hasn't been already.
263 if not expires:
264 self.cookies[key]["expires"] = http_date(time.time() + max_age)
265 if path is not None:
266 self.cookies[key]["path"] = path
267 if domain is not None:
268 self.cookies[key]["domain"] = domain
269 if secure:
270 self.cookies[key]["secure"] = True
271 if httponly:
272 self.cookies[key]["httponly"] = True
273 if samesite:
274 if samesite.lower() not in ("lax", "none", "strict"):
275 raise ValueError('samesite must be "lax", "none", or "strict".')
276 self.cookies[key]["samesite"] = samesite
278 def setdefault(self, key, value):
279 """Set a header unless it has already been set."""
280 self.headers.setdefault(key, value)
282 def set_signed_cookie(self, key, value, salt="", **kwargs):
283 value = signing.get_cookie_signer(salt=key + salt).sign(value)
284 return self.set_cookie(key, value, **kwargs)
286 def delete_cookie(self, key, path="/", domain=None, samesite=None):
287 # Browsers can ignore the Set-Cookie header if the cookie doesn't use
288 # the secure flag and:
289 # - the cookie name starts with "__Host-" or "__Secure-", or
290 # - the samesite is "none".
291 secure = key.startswith(("__Secure-", "__Host-")) or (
292 samesite and samesite.lower() == "none"
293 )
294 self.set_cookie(
295 key,
296 max_age=0,
297 path=path,
298 domain=domain,
299 secure=secure,
300 expires="Thu, 01 Jan 1970 00:00:00 GMT",
301 samesite=samesite,
302 )
304 # Common methods used by subclasses
306 def make_bytes(self, value):
307 """Turn a value into a bytestring encoded in the output charset."""
308 # Per PEP 3333, this response body must be bytes. To avoid returning
309 # an instance of a subclass, this function returns `bytes(value)`.
310 # This doesn't make a copy when `value` already contains bytes.
312 # Handle string types -- we can't rely on force_bytes here because:
313 # - Python attempts str conversion first
314 # - when self._charset != 'utf-8' it re-encodes the content
315 if isinstance(value, (bytes, memoryview)):
316 return bytes(value)
317 if isinstance(value, str):
318 return bytes(value.encode(self.charset))
319 # Handle non-string types.
320 return str(value).encode(self.charset)
322 # These methods partially implement the file-like object interface.
323 # See https://docs.python.org/library/io.html#io.IOBase
325 # The WSGI server must call this method upon completion of the request.
326 # See http://blog.dscpl.com.au/2012/10/obligations-for-calling-close-on.html
327 def close(self):
328 for closer in self._resource_closers:
329 try:
330 closer()
331 except Exception:
332 pass
333 # Free resources that were still referenced.
334 self._resource_closers.clear()
335 self.closed = True
336 signals.request_finished.send(sender=self._handler_class)
338 def write(self, content):
339 raise OSError("This %s instance is not writable" % self.__class__.__name__)
341 def flush(self):
342 pass
344 def tell(self):
345 raise OSError(
346 "This %s instance cannot tell its position" % self.__class__.__name__
347 )
349 # These methods partially implement a stream-like object interface.
350 # See https://docs.python.org/library/io.html#io.IOBase
352 def readable(self):
353 return False
355 def seekable(self):
356 return False
358 def writable(self):
359 return False
361 def writelines(self, lines):
362 raise OSError("This %s instance is not writable" % self.__class__.__name__)
365class HttpResponse(HttpResponseBase):
366 """
367 An HTTP response class with a string as content.
369 This content can be read, appended to, or replaced.
370 """
372 streaming = False
374 def __init__(self, content=b"", *args, **kwargs):
375 super().__init__(*args, **kwargs)
376 # Content is a bytestring. See the `content` property methods.
377 self.content = content
379 def __repr__(self):
380 return "<%(cls)s status_code=%(status_code)d%(content_type)s>" % {
381 "cls": self.__class__.__name__,
382 "status_code": self.status_code,
383 "content_type": self._content_type_for_repr,
384 }
386 def serialize(self):
387 """Full HTTP message, including headers, as a bytestring."""
388 return self.serialize_headers() + b"\r\n\r\n" + self.content
390 __bytes__ = serialize
392 @property
393 def content(self):
394 return b"".join(self._container)
396 @content.setter
397 def content(self, value):
398 # Consume iterators upon assignment to allow repeated iteration.
399 if hasattr(value, "__iter__") and not isinstance(
400 value, (bytes, memoryview, str)
401 ):
402 content = b"".join(self.make_bytes(chunk) for chunk in value)
403 if hasattr(value, "close"):
404 try:
405 value.close()
406 except Exception:
407 pass
408 else:
409 content = self.make_bytes(value)
410 # Create a list of properly encoded bytestrings to support write().
411 self._container = [content]
412 self.__dict__.pop("text", None)
414 @cached_property
415 def text(self):
416 return self.content.decode(self.charset or "utf-8")
418 def __iter__(self):
419 return iter(self._container)
421 def write(self, content):
422 self._container.append(self.make_bytes(content))
424 def tell(self):
425 return len(self.content)
427 def getvalue(self):
428 return self.content
430 def writable(self):
431 return True
433 def writelines(self, lines):
434 for line in lines:
435 self.write(line)
438class StreamingHttpResponse(HttpResponseBase):
439 """
440 A streaming HTTP response class with an iterator as content.
442 This should only be iterated once, when the response is streamed to the
443 client. However, it can be appended to or replaced with a new iterator
444 that wraps the original content (or yields entirely new content).
445 """
447 streaming = True
449 def __init__(self, streaming_content=(), *args, **kwargs):
450 super().__init__(*args, **kwargs)
451 # `streaming_content` should be an iterable of bytestrings.
452 # See the `streaming_content` property methods.
453 self.streaming_content = streaming_content
455 def __repr__(self):
456 return "<%(cls)s status_code=%(status_code)d%(content_type)s>" % {
457 "cls": self.__class__.__qualname__,
458 "status_code": self.status_code,
459 "content_type": self._content_type_for_repr,
460 }
462 @property
463 def content(self):
464 raise AttributeError(
465 "This %s instance has no `content` attribute. Use "
466 "`streaming_content` instead." % self.__class__.__name__
467 )
469 @property
470 def text(self):
471 raise AttributeError(
472 "This %s instance has no `text` attribute." % self.__class__.__name__
473 )
475 @property
476 def streaming_content(self):
477 if self.is_async:
478 # pull to lexical scope to capture fixed reference in case
479 # streaming_content is set again later.
480 _iterator = self._iterator
482 async def awrapper():
483 async for part in _iterator:
484 yield self.make_bytes(part)
486 return awrapper()
487 else:
488 return map(self.make_bytes, self._iterator)
490 @streaming_content.setter
491 def streaming_content(self, value):
492 self._set_streaming_content(value)
494 def _set_streaming_content(self, value):
495 # Ensure we can never iterate on "value" more than once.
496 try:
497 self._iterator = iter(value)
498 self.is_async = False
499 except TypeError:
500 self._iterator = aiter(value)
501 self.is_async = True
502 if hasattr(value, "close"):
503 self._resource_closers.append(value.close)
505 def __iter__(self):
506 try:
507 return iter(self.streaming_content)
508 except TypeError:
509 warnings.warn(
510 "StreamingHttpResponse must consume asynchronous iterators in order to "
511 "serve them synchronously. Use a synchronous iterator instead.",
512 Warning,
513 stacklevel=2,
514 )
516 # async iterator. Consume in async_to_sync and map back.
517 async def to_list(_iterator):
518 as_list = []
519 async for chunk in _iterator:
520 as_list.append(chunk)
521 return as_list
523 return map(self.make_bytes, iter(async_to_sync(to_list)(self._iterator)))
525 async def __aiter__(self):
526 try:
527 async for part in self.streaming_content:
528 yield part
529 except TypeError:
530 warnings.warn(
531 "StreamingHttpResponse must consume synchronous iterators in order to "
532 "serve them asynchronously. Use an asynchronous iterator instead.",
533 Warning,
534 stacklevel=2,
535 )
536 # sync iterator. Consume via sync_to_async and yield via async
537 # generator.
538 for part in await sync_to_async(list)(self.streaming_content):
539 yield part
541 def getvalue(self):
542 return b"".join(self.streaming_content)
545class FileResponse(StreamingHttpResponse):
546 """
547 A streaming HTTP response class optimized for files.
548 """
550 block_size = 4096
552 def __init__(self, *args, as_attachment=False, filename="", **kwargs):
553 self.as_attachment = as_attachment
554 self.filename = filename
555 self._no_explicit_content_type = (
556 "content_type" not in kwargs or kwargs["content_type"] is None
557 )
558 super().__init__(*args, **kwargs)
560 def _set_streaming_content(self, value):
561 if not hasattr(value, "read"):
562 self.file_to_stream = None
563 return super()._set_streaming_content(value)
565 self.file_to_stream = filelike = value
566 if hasattr(filelike, "close"):
567 self._resource_closers.append(filelike.close)
568 value = iter(lambda: filelike.read(self.block_size), b"")
569 self.set_headers(filelike)
570 super()._set_streaming_content(value)
572 def set_headers(self, filelike):
573 """
574 Set some common response headers (Content-Length, Content-Type, and
575 Content-Disposition) based on the `filelike` response content.
576 """
577 filename = getattr(filelike, "name", "")
578 filename = filename if isinstance(filename, str) else ""
579 seekable = hasattr(filelike, "seek") and (
580 not hasattr(filelike, "seekable") or filelike.seekable()
581 )
582 if hasattr(filelike, "tell"):
583 if seekable:
584 initial_position = filelike.tell()
585 filelike.seek(0, io.SEEK_END)
586 self.headers["Content-Length"] = filelike.tell() - initial_position
587 filelike.seek(initial_position)
588 elif hasattr(filelike, "getbuffer"):
589 self.headers["Content-Length"] = (
590 filelike.getbuffer().nbytes - filelike.tell()
591 )
592 elif os.path.exists(filename):
593 self.headers["Content-Length"] = (
594 os.path.getsize(filename) - filelike.tell()
595 )
596 elif seekable:
597 self.headers["Content-Length"] = sum(
598 iter(lambda: len(filelike.read(self.block_size)), 0)
599 )
600 filelike.seek(-int(self.headers["Content-Length"]), io.SEEK_END)
602 filename = os.path.basename(self.filename or filename)
603 if self._no_explicit_content_type:
604 if filename:
605 content_type, encoding = mimetypes.guess_type(filename)
606 # Encoding isn't set to prevent browsers from automatically
607 # uncompressing files.
608 content_type = {
609 "br": "application/x-brotli",
610 "bzip2": "application/x-bzip",
611 "compress": "application/x-compress",
612 "gzip": "application/gzip",
613 "xz": "application/x-xz",
614 }.get(encoding, content_type)
615 self.headers["Content-Type"] = (
616 content_type or "application/octet-stream"
617 )
618 else:
619 self.headers["Content-Type"] = "application/octet-stream"
621 if content_disposition := content_disposition_header(
622 self.as_attachment, filename
623 ):
624 self.headers["Content-Disposition"] = content_disposition
627class HttpResponseRedirectBase(HttpResponse):
628 allowed_schemes = ["http", "https", "ftp"]
630 def __init__(self, redirect_to, preserve_request=False, *args, **kwargs):
631 super().__init__(*args, **kwargs)
632 self["Location"] = iri_to_uri(redirect_to)
633 parsed = urlsplit(str(redirect_to))
634 if preserve_request:
635 self.status_code = self.status_code_preserve_request
636 if parsed.scheme and parsed.scheme not in self.allowed_schemes:
637 raise DisallowedRedirect(
638 "Unsafe redirect to URL with protocol '%s'" % parsed.scheme
639 )
641 url = property(lambda self: self["Location"])
643 def __repr__(self):
644 return (
645 '<%(cls)s status_code=%(status_code)d%(content_type)s, url="%(url)s">'
646 % {
647 "cls": self.__class__.__name__,
648 "status_code": self.status_code,
649 "content_type": self._content_type_for_repr,
650 "url": self.url,
651 }
652 )
655class HttpResponseRedirect(HttpResponseRedirectBase):
656 status_code = 302
657 status_code_preserve_request = 307
660class HttpResponsePermanentRedirect(HttpResponseRedirectBase):
661 status_code = 301
662 status_code_preserve_request = 308
665class HttpResponseNotModified(HttpResponse):
666 status_code = 304
668 def __init__(self, *args, **kwargs):
669 super().__init__(*args, **kwargs)
670 del self["content-type"]
672 @HttpResponse.content.setter
673 def content(self, value):
674 if value:
675 raise AttributeError(
676 "You cannot set content to a 304 (Not Modified) response"
677 )
678 self._container = []
681class HttpResponseBadRequest(HttpResponse):
682 status_code = 400
685class HttpResponseNotFound(HttpResponse):
686 status_code = 404
689class HttpResponseForbidden(HttpResponse):
690 status_code = 403
693class HttpResponseNotAllowed(HttpResponse):
694 status_code = 405
696 def __init__(self, permitted_methods, *args, **kwargs):
697 super().__init__(*args, **kwargs)
698 self["Allow"] = ", ".join(permitted_methods)
700 def __repr__(self):
701 return "<%(cls)s [%(methods)s] status_code=%(status_code)d%(content_type)s>" % {
702 "cls": self.__class__.__name__,
703 "status_code": self.status_code,
704 "content_type": self._content_type_for_repr,
705 "methods": self["Allow"],
706 }
709class HttpResponseGone(HttpResponse):
710 status_code = 410
713class HttpResponseServerError(HttpResponse):
714 status_code = 500
717class Http404(Exception):
718 pass
721class JsonResponse(HttpResponse):
722 """
723 An HTTP response class that consumes data to be serialized to JSON.
725 :param data: Data to be dumped into json. By default only ``dict`` objects
726 are allowed to be passed due to a security flaw before ECMAScript 5. See
727 the ``safe`` parameter for more information.
728 :param encoder: Should be a json encoder class. Defaults to
729 ``django.core.serializers.json.DjangoJSONEncoder``.
730 :param safe: Controls if only ``dict`` objects may be serialized. Defaults
731 to ``True``.
732 :param json_dumps_params: A dictionary of kwargs passed to json.dumps().
733 """
735 def __init__(
736 self,
737 data,
738 encoder=DjangoJSONEncoder,
739 safe=True,
740 json_dumps_params=None,
741 **kwargs,
742 ):
743 if safe and not isinstance(data, dict):
744 raise TypeError(
745 "In order to allow non-dict objects to be serialized set the "
746 "safe parameter to False."
747 )
748 if json_dumps_params is None:
749 json_dumps_params = {}
750 kwargs.setdefault("content_type", "application/json")
751 data = json.dumps(data, cls=encoder, **json_dumps_params)
752 super().__init__(content=data, **kwargs)