1from __future__ import annotations
2
3import datetime
4import io
5import json
6import mimetypes
7import os
8import re
9import sys
10import time
11from collections.abc import AsyncIterator, Iterator
12from email.header import Header
13from http.client import responses
14from http.cookies import SimpleCookie
15from typing import IO, Any
16
17from plain.http.cookie import sign_cookie_value
18from plain.json import PlainJSONEncoder
19from plain.utils import timezone
20from plain.utils.datastructures import CaseInsensitiveMapping
21from plain.utils.encoding import iri_to_uri
22from plain.utils.http import content_disposition_header, http_date
23from plain.utils.regex_helper import _lazy_re_compile
24
25_charset_from_content_type_re = _lazy_re_compile(
26 r";\s*charset=(?P<charset>[^\s;]+)", re.IGNORECASE
27)
28
29
30class ResponseHeaders(CaseInsensitiveMapping):
31 def __init__(self, data: dict[str, Any] | None = None):
32 """
33 Populate the initial data using __setitem__ to ensure values are
34 correctly encoded.
35 """
36 self._store = {}
37 if data:
38 for header, value in self._unpack_items(data):
39 self[header] = value
40
41 def _convert_to_charset(
42 self, value: str | bytes, charset: str, mime_encode: bool = False
43 ) -> str:
44 """
45 Convert headers key/value to ascii/latin-1 native strings.
46 `charset` must be 'ascii' or 'latin-1'. If `mime_encode` is True and
47 `value` can't be represented in the given charset, apply MIME-encoding.
48 """
49 try:
50 if isinstance(value, str):
51 # Ensure string is valid in given charset
52 value.encode(charset)
53 elif isinstance(value, bytes):
54 # Convert bytestring using given charset
55 value = value.decode(charset)
56 else:
57 value = str(value)
58 # Ensure string is valid in given charset.
59 value.encode(charset)
60 if "\n" in value or "\r" in value:
61 raise BadHeaderError(
62 f"Header values can't contain newlines (got {value!r})"
63 )
64 except UnicodeError as e:
65 # Encoding to a string of the specified charset failed, but we
66 # don't know what type that value was, or if it contains newlines,
67 # which we may need to check for before sending it to be
68 # encoded for multiple character sets.
69 if (isinstance(value, bytes) and (b"\n" in value or b"\r" in value)) or (
70 isinstance(value, str) and ("\n" in value or "\r" in value)
71 ):
72 raise BadHeaderError(
73 f"Header values can't contain newlines (got {value!r})"
74 ) from e
75 if mime_encode:
76 value = Header(value, "utf-8", maxlinelen=sys.maxsize).encode()
77 else:
78 if isinstance(
79 e, UnicodeDecodeError | UnicodeEncodeError | UnicodeTranslateError
80 ):
81 e.reason += f", HTTP response headers must be in {charset} format"
82 raise
83 return value
84
85 def __delitem__(self, key: str) -> None:
86 self.pop(key)
87
88 def __setitem__(self, key: str, value: str | bytes | None) -> None:
89 key = self._convert_to_charset(key, "ascii")
90 if value is None:
91 self._store[key.lower()] = (key, None)
92 else:
93 value = self._convert_to_charset(value, "latin-1", mime_encode=True)
94 self._store[key.lower()] = (key, value)
95
96 def pop(self, key: str, default: Any = None) -> Any:
97 return self._store.pop(key.lower(), default)
98
99 def setdefault(self, key: str, value: str | bytes) -> None:
100 if key not in self:
101 self[key] = value
102
103
104class BadHeaderError(ValueError):
105 pass
106
107
108# Distinguishes "subclass declared status_code = None" (invalid, rejected)
109# from "subclass declared nothing".
110_NOT_DECLARED: Any = object()
111
112# Private sentinel streaming subclasses pass to skip bytes-body setup in
113# Response.__init__ entirely — their `content` property raises, so the
114# setter (and _container) must never be touched. A dedicated object, not
115# None: an explicit None is a real value meaning "no body".
116_NO_CONTENT: Any = object()
117
118
119def is_valid_status_code(status_code: object) -> bool:
120 """True for an int in the constructible range (bools excluded)."""
121 return (
122 isinstance(status_code, int)
123 and not isinstance(status_code, bool)
124 and 200 <= status_code <= 599
125 )
126
127
128def status_omits_body(status_code: int | None) -> bool:
129 """True for statuses whose responses never have a body (RFC 9110).
130
131 A client stops reading a 1xx/204/304 response at the header block,
132 so any body bytes written after it would be parsed as the start of
133 the NEXT response on a keep-alive connection. The single definition
134 shared by Response construction, the test client, and the server's
135 h1/h2 writers. (The 1xx arm is a wire-level backstop: 1xx is
136 unrepresentable on a Response — rejected at construction, and
137 status_code has no setter.)
138 """
139 return status_code is not None and (status_code < 200 or status_code in (204, 304))
140
141
142def response_omits_body(*, method: str | None, status_code: int | None) -> bool:
143 """True when a response sends only headers: HEAD, or a bodiless status."""
144 return method == "HEAD" or status_omits_body(status_code)
145
146
147def content_length_forbidden(status_code: int | None) -> bool:
148 """RFC 9110 8.6: 1xx and 204 must not carry Content-Length; 304 may."""
149 return status_omits_body(status_code) and status_code != 304
150
151
152class Response:
153 """
154 An HTTP response class with a bytes body.
155
156 Base class for all response types — streaming variants subclass this
157 and swap the body for an iterator. Users annotate handler returns and
158 middleware with `Response` to cover all response shapes.
159 """
160
161 streaming = False
162 _default_status_code = 200
163
164 def __init_subclass__(cls, **kwargs: object) -> None:
165 super().__init_subclass__(**kwargs)
166 declared = cls.__dict__.get("status_code", _NOT_DECLARED)
167 if declared is _NOT_DECLARED or isinstance(declared, property):
168 return
169 # Declarative subclasses write `status_code = 304`. Left as a
170 # plain class attribute it would shadow the validating property
171 # below, so validate it at the line that wrote it and fold it
172 # into the default the property serves.
173 if not is_valid_status_code(declared):
174 raise ValueError(
175 f"{cls.__name__}.status_code must be an integer from "
176 f"200 to 599, got {declared!r}."
177 )
178 delattr(cls, "status_code")
179 cls._default_status_code = declared
180
181 @property
182 def status_code(self) -> int:
183 """Fixed at construction — there is deliberately no setter.
184
185 Status is part of a response's identity: headers defaulting,
186 the bodiless rules (RFC 9110), transports, and caches all read
187 it as a settled fact. Pass `status_code=` to the constructor
188 (`TemplateView.render()` takes it too) instead of mutating a
189 built response — assignment fails the type check and raises
190 AttributeError at runtime.
191 """
192 return self._status_code
193
194 def __init__(
195 self,
196 content: bytes | str | Iterator[bytes] | None = b"",
197 *,
198 content_type: str | None = None,
199 status_code: int | None = None,
200 reason: str | None = None,
201 charset: str | None = None,
202 headers: dict[str, Any] | None = None,
203 ):
204 self.headers = ResponseHeaders(headers)
205 self._charset = charset
206 # Materialized on every instance so copies (e.g. the test
207 # client's) never depend on class lookup. The class default was
208 # validated at definition; an argument is validated here — the
209 # only door, since status_code has no setter.
210 if status_code is None:
211 self._status_code = self._default_status_code
212 else:
213 try:
214 status_code = int(status_code)
215 except (ValueError, TypeError):
216 raise TypeError("HTTP status code must be an integer.")
217 if not is_valid_status_code(status_code):
218 raise ValueError(
219 "HTTP status code must be an integer from 200 to 599 "
220 "(1xx interim responses are sent by the server, not "
221 "application code)."
222 )
223 self._status_code = status_code
224 if content is _NO_CONTENT and status_omits_body(self._status_code):
225 # Streaming subclasses pass the sentinel. Refusing here —
226 # before the iterator is ever assigned — means the caller
227 # keeps ownership of it and nothing needs closing.
228 raise ValueError(
229 f"A {self._status_code} response cannot have a body — "
230 "it can't be a streaming response."
231 )
232 if "Content-Type" not in self.headers:
233 # A bodiless status (204/304) gets no default Content-Type:
234 # there is no representation to describe, and on a 304 caches
235 # update stored representation headers from the response
236 # (RFC 9110 15.4.5). An explicit content_type is respected.
237 if content_type is None and not status_omits_body(self.status_code):
238 content_type = f"text/html; charset={self.charset}"
239 if content_type is not None:
240 self.headers["Content-Type"] = content_type
241 elif content_type:
242 raise ValueError(
243 "'headers' must not contain 'Content-Type' when the "
244 "'content_type' parameter is provided."
245 )
246 self._resource_closers = []
247 self.cookies = SimpleCookie()
248 self.closed = False
249 self._reason_phrase = reason
250 # Exception that caused this response, if any (primarily for 500 errors)
251 self.exception: Exception | None = None
252 # Whether the server should log this response in the access log
253 self.log_access: bool = True
254 if content is not _NO_CONTENT:
255 self.content = content
256
257 @property
258 def reason_phrase(self) -> str:
259 if self._reason_phrase is not None:
260 return self._reason_phrase
261 # Leave self._reason_phrase unset in order to use the default
262 # reason phrase for status code.
263 return responses.get(self.status_code, "Unknown Status Code")
264
265 @reason_phrase.setter
266 def reason_phrase(self, value: str) -> None:
267 self._reason_phrase = value
268
269 @property
270 def charset(self) -> str:
271 if self._charset is not None:
272 return self._charset
273 # The Content-Type header may not yet be set, because the charset is
274 # being inserted *into* it.
275 if (content_type := self.headers.get("Content-Type")) and (
276 matched := _charset_from_content_type_re.search(content_type)
277 ):
278 # Extract the charset and strip its double quotes.
279 # Note that having parsed it from the Content-Type, we don't
280 # store it back into the _charset for later intentionally, to
281 # allow for the Content-Type to be switched again later.
282 return matched["charset"].replace('"', "")
283 return "utf-8"
284
285 @charset.setter
286 def charset(self, value: str) -> None:
287 self._charset = value
288
289 @property
290 def _content_type_for_repr(self) -> str:
291 return (
292 ', "{}"'.format(self.headers["Content-Type"])
293 if "Content-Type" in self.headers
294 else ""
295 )
296
297 def set_cookie(
298 self,
299 key: str,
300 value: str = "",
301 max_age: float | datetime.timedelta | None = None,
302 expires: str | datetime.datetime | None = None,
303 path: str | None = "/",
304 domain: str | None = None,
305 secure: bool = False,
306 httponly: bool = False,
307 samesite: str | None = None,
308 ) -> None:
309 """
310 Set a cookie.
311
312 ``expires`` can be:
313 - a string in the correct format,
314 - a naive ``datetime.datetime`` object in UTC,
315 - an aware ``datetime.datetime`` object in any time zone.
316 If it is a ``datetime.datetime`` object then calculate ``max_age``.
317
318 ``max_age`` can be:
319 - int/float specifying seconds,
320 - ``datetime.timedelta`` object.
321 """
322 self.cookies[key] = value
323 if expires is not None:
324 if isinstance(expires, datetime.datetime):
325 if timezone.is_naive(expires):
326 expires = timezone.make_aware(expires, datetime.UTC)
327 delta = expires - datetime.datetime.now(tz=datetime.UTC)
328 # Add one second so the date matches exactly (a fraction of
329 # time gets lost between converting to a timedelta and
330 # then the date string).
331 delta += datetime.timedelta(seconds=1)
332 # Just set max_age - the max_age logic will set expires.
333 expires = None
334 if max_age is not None:
335 raise ValueError("'expires' and 'max_age' can't be used together.")
336 max_age = max(0, delta.days * 86400 + delta.seconds)
337 else:
338 self.cookies[key]["expires"] = expires
339 else:
340 self.cookies[key]["expires"] = ""
341 if max_age is not None:
342 if isinstance(max_age, datetime.timedelta):
343 max_age = max_age.total_seconds()
344 self.cookies[key]["max-age"] = int(max_age)
345 # IE requires expires, so set it if hasn't been already.
346 if not expires:
347 self.cookies[key]["expires"] = http_date(time.time() + max_age)
348 if path is not None:
349 self.cookies[key]["path"] = path
350 if domain is not None:
351 self.cookies[key]["domain"] = domain
352 if secure:
353 self.cookies[key]["secure"] = True
354 if httponly:
355 self.cookies[key]["httponly"] = True
356 if samesite:
357 if samesite.lower() not in ("lax", "none", "strict"):
358 raise ValueError('samesite must be "lax", "none", or "strict".')
359 self.cookies[key]["samesite"] = samesite
360
361 def set_signed_cookie(
362 self, key: str, value: str, salt: str = "", **kwargs: Any
363 ) -> None:
364 """Set a cookie signed with the SECRET_KEY."""
365
366 signed_value = sign_cookie_value(key, value, salt)
367 return self.set_cookie(key, signed_value, **kwargs)
368
369 def delete_cookie(
370 self,
371 key: str,
372 path: str = "/",
373 domain: str | None = None,
374 samesite: str | None = None,
375 ) -> None:
376 # Browsers can ignore the Set-Cookie header if the cookie doesn't use
377 # the secure flag and:
378 # - the cookie name starts with "__Host-" or "__Secure-", or
379 # - the samesite is "none".
380 secure = key.startswith(("__Secure-", "__Host-")) or bool(
381 samesite and samesite.lower() == "none"
382 )
383 self.set_cookie(
384 key,
385 max_age=0,
386 path=path,
387 domain=domain,
388 secure=secure,
389 expires="Thu, 01 Jan 1970 00:00:00 GMT",
390 samesite=samesite,
391 )
392
393 # Common methods used by subclasses
394
395 def make_bytes(self, value: str | bytes) -> bytes:
396 """Turn a value into a bytestring encoded in the output charset."""
397 # Per PEP 3333, this response body must be bytes. To avoid returning
398 # an instance of a subclass, this function returns `bytes(value)`.
399 # This doesn't make a copy when `value` already contains bytes.
400
401 # Handle string types -- we can't rely on force_bytes here because:
402 # - Python attempts str conversion first
403 # - when self._charset != 'utf-8' it re-encodes the content
404 if isinstance(value, bytes | memoryview):
405 return bytes(value)
406 if isinstance(value, str):
407 return bytes(value.encode(self.charset))
408 # Handle non-string types.
409 return str(value).encode(self.charset)
410
411 # The server must call this method upon completion of the request.
412 # See http://blog.dscpl.com.au/2012/10/obligations-for-calling-close-on.html
413 def close(self) -> None:
414 if self.closed:
415 return
416 for closer in self._resource_closers:
417 try:
418 closer()
419 except Exception:
420 pass
421 # Free resources that were still referenced.
422 self._resource_closers.clear()
423 self.closed = True
424
425 def __repr__(self) -> str:
426 return "<%(cls)s status_code=%(status_code)d%(content_type)s>" % { # noqa: UP031
427 "cls": self.__class__.__name__,
428 "status_code": self.status_code,
429 "content_type": self._content_type_for_repr,
430 }
431
432 @property
433 def content(self) -> bytes:
434 return b"".join(self._container)
435
436 @content.setter
437 def content(self, value: bytes | str | Iterator[bytes] | None) -> None:
438 # Consume iterators upon assignment to allow repeated iteration.
439 if hasattr(value, "__iter__") and not isinstance(
440 value, bytes | memoryview | str
441 ):
442 content = b"".join(self.make_bytes(chunk) for chunk in value)
443 if hasattr(value, "close") and callable(getattr(value, "close")):
444 try:
445 value.close() # ty: ignore[call-non-callable]
446 except Exception:
447 pass
448 elif value is None:
449 # An explicit None means "no body".
450 content = b""
451 else:
452 content = self.make_bytes(value)
453 if content and status_omits_body(self.status_code):
454 raise ValueError(
455 f"A {self.status_code} response cannot have a body — "
456 "send the content with a 200, or drop it."
457 )
458 self._container = [content]
459
460 def __iter__(self) -> Iterator[bytes]:
461 return iter(self._container)
462
463
464class StreamingResponse(Response):
465 """
466 A streaming HTTP response class with an iterator as content.
467
468 This should only be iterated once, when the response is streamed to the
469 client. However, it can be appended to or replaced with a new iterator
470 that wraps the original content (or yields entirely new content).
471 """
472
473 streaming = True
474
475 def __init__(
476 self,
477 streaming_content: Any = (),
478 *,
479 content_type: str | None = None,
480 status_code: int | None = None,
481 reason: str | None = None,
482 charset: str | None = None,
483 headers: dict[str, Any] | None = None,
484 ):
485 super().__init__(
486 content=_NO_CONTENT,
487 content_type=content_type,
488 status_code=status_code,
489 reason=reason,
490 charset=charset,
491 headers=headers,
492 )
493 # `streaming_content` should be an iterable of bytestrings.
494 # See the `streaming_content` property methods.
495 self.streaming_content = streaming_content
496
497 @property
498 def content(self) -> bytes:
499 raise AttributeError(
500 f"This {self.__class__.__name__} instance has no `content` attribute. Use "
501 "`streaming_content` instead."
502 )
503
504 @property
505 def streaming_content(self) -> Iterator[bytes]:
506 return map(self.make_bytes, self._iterator)
507
508 @streaming_content.setter
509 def streaming_content(self, value: Iterator[bytes | str]) -> None:
510 self._set_streaming_content(value)
511
512 def _set_streaming_content(self, value: Iterator[bytes | str]) -> None:
513 # Ensure we can never iterate on "value" more than once.
514 self._iterator = iter(value)
515 if hasattr(value, "close"):
516 self._resource_closers.append(value.close)
517
518 def __iter__(self) -> Iterator[bytes]:
519 return iter(self.streaming_content)
520
521
522class AsyncStreamingResponse(Response):
523 """
524 A streaming HTTP response class with an async iterator as content.
525
526 Used for long-lived connections like Server-Sent Events (SSE) where
527 data arrives asynchronously and should be streamed to the client
528 without buffering the entire response.
529 """
530
531 streaming = True
532
533 def __init__(
534 self,
535 streaming_content: AsyncIterator[bytes | str],
536 *,
537 content_type: str | None = None,
538 status_code: int | None = None,
539 reason: str | None = None,
540 charset: str | None = None,
541 headers: dict[str, Any] | None = None,
542 ):
543 super().__init__(
544 content=_NO_CONTENT,
545 content_type=content_type,
546 status_code=status_code,
547 reason=reason,
548 charset=charset,
549 headers=headers,
550 )
551 self._async_iterator = streaming_content
552
553 @property
554 def content(self) -> bytes:
555 raise AttributeError(
556 f"This {self.__class__.__name__} instance has no `content` attribute. Use "
557 "`streaming_content` instead."
558 )
559
560 def _to_buffered_response(self, body: bytes) -> Response:
561 """Materialize the streamed body into a plain Response.
562
563 Used by the test client after collecting the stream. The body
564 routes through the constructor (and its validation), then the
565 rest of the instance state — including anything app code set on
566 the response — transfers wholesale, so tests assert against the
567 same object shape production sends. `closed` deliberately starts
568 fresh, and the resource closers move over.
569 """
570 response = Response(body, status_code=self.status_code)
571 state = {
572 k: v
573 for k, v in self.__dict__.items()
574 if k not in ("_async_iterator", "closed", "_container", "_status_code")
575 }
576 response.__dict__.update(state)
577 self._resource_closers = []
578 return response
579
580 def __iter__(self) -> Iterator[bytes]:
581 raise TypeError(
582 f"{self.__class__.__name__} is async — use `async for` / `__aiter__` instead."
583 )
584
585 async def __aiter__(self) -> AsyncIterator[bytes]:
586 async for chunk in self._async_iterator:
587 yield self.make_bytes(chunk)
588
589 async def aclose(self) -> None:
590 """Close the underlying async iterator if it supports it."""
591 close = getattr(self._async_iterator, "aclose", None)
592 if close is not None:
593 await close()
594
595
596class FileResponse(StreamingResponse):
597 """
598 A streaming HTTP response class optimized for files.
599 """
600
601 block_size = 4096
602
603 def __init__(
604 self,
605 streaming_content: Any = (),
606 *,
607 as_attachment: bool = False,
608 filename: str = "",
609 content_type: str | None = None,
610 status_code: int | None = None,
611 reason: str | None = None,
612 charset: str | None = None,
613 headers: dict[str, Any] | None = None,
614 ):
615 self.as_attachment = as_attachment
616 self.filename = filename
617 self._no_explicit_content_type = content_type is None
618 try:
619 super().__init__(
620 streaming_content,
621 content_type=content_type,
622 status_code=status_code,
623 reason=reason,
624 charset=charset,
625 headers=headers,
626 )
627 except ValueError:
628 # Unlike a generic iterator (which stays the caller's on a
629 # bodiless rejection), FileResponse owns the file handle it
630 # was given — the idiomatic call is FileResponse(open(p)),
631 # which leaves the caller nothing to close.
632 close = getattr(streaming_content, "close", None)
633 if callable(close):
634 close()
635 raise
636
637 def _set_streaming_content(self, value: Any) -> None:
638 if not hasattr(value, "read"):
639 self.file_to_stream = None
640 return super()._set_streaming_content(value)
641
642 self.file_to_stream = filelike = value
643 if hasattr(filelike, "close"):
644 self._resource_closers.append(filelike.close)
645 value = iter(lambda: filelike.read(self.block_size), b"")
646 self.set_headers(filelike)
647 super()._set_streaming_content(value)
648
649 def set_headers(self, filelike: IO[bytes]) -> None:
650 """
651 Set some common response headers (Content-Length, Content-Type, and
652 Content-Disposition) based on the `filelike` response content.
653 """
654 filename = getattr(filelike, "name", "")
655 filename = filename if isinstance(filename, str) else ""
656 seekable = hasattr(filelike, "seek") and (
657 not hasattr(filelike, "seekable") or filelike.seekable()
658 )
659 if hasattr(filelike, "tell"):
660 if seekable:
661 initial_position = filelike.tell()
662 filelike.seek(0, io.SEEK_END)
663 self.headers["Content-Length"] = str(filelike.tell() - initial_position)
664 filelike.seek(initial_position)
665 elif callable(getbuffer := getattr(filelike, "getbuffer", None)):
666 self.headers["Content-Length"] = str(
667 getbuffer().nbytes - filelike.tell()
668 )
669 elif os.path.exists(filename):
670 self.headers["Content-Length"] = str(
671 os.path.getsize(filename) - filelike.tell()
672 )
673 elif seekable:
674 self.headers["Content-Length"] = str(
675 sum(iter(lambda: len(filelike.read(self.block_size)), 0))
676 )
677 filelike.seek(-int(self.headers["Content-Length"]), io.SEEK_END)
678
679 filename = os.path.basename(self.filename or filename)
680 if self._no_explicit_content_type:
681 if filename:
682 content_type, encoding = mimetypes.guess_type(filename)
683 # Encoding isn't set to prevent browsers from automatically
684 # uncompressing files.
685 encoding_types: dict[str, str] = {
686 "br": "application/x-brotli",
687 "bzip2": "application/x-bzip",
688 "compress": "application/x-compress",
689 "gzip": "application/gzip",
690 "xz": "application/x-xz",
691 }
692 if encoding and encoding in encoding_types:
693 content_type = encoding_types[encoding]
694 self.headers["Content-Type"] = (
695 content_type or "application/octet-stream"
696 )
697 else:
698 self.headers["Content-Type"] = "application/octet-stream"
699
700 if content_disposition := content_disposition_header(
701 self.as_attachment, filename
702 ):
703 self.headers["Content-Disposition"] = content_disposition
704
705
706# A URI scheme per RFC 3986: a letter, then letters, digits, "+", "-", ".".
707_SCHEME_PREFIX_RE = re.compile(r"[a-zA-Z][a-zA-Z0-9+.\-]*:")
708
709
710def _is_external_url(url: str) -> bool:
711 """Check if a URL would redirect to an external host."""
712 if not url:
713 return False
714 # Browsers strip leading whitespace from Location headers
715 url = url.strip()
716 # Browsers normalize backslashes to forward slashes in URLs,
717 # so \\ and /\ are equivalent to //
718 if url[:2].replace("\\", "/") == "//":
719 return True
720 # Any scheme sends the browser off this origin, with or without "//"
721 # after it. Browsers normalize "http:/evil.com" (a single slash) to
722 # "http://evil.com", so matching on "://" alone lets that through.
723 return _SCHEME_PREFIX_RE.match(url) is not None
724
725
726class RedirectResponse(Response):
727 """HTTP redirect response"""
728
729 def __init__(
730 self,
731 redirect_to: str,
732 *,
733 status_code: int,
734 allow_external: bool = False,
735 content_type: str | None = None,
736 reason: str | None = None,
737 charset: str | None = None,
738 headers: dict[str, Any] | None = None,
739 ):
740 if not allow_external and _is_external_url(redirect_to):
741 raise ValueError(
742 f"Unsafe redirect URL: {redirect_to!r}. "
743 "RedirectResponse does not allow external URLs by default. "
744 "Use allow_external=True if you intentionally want to redirect "
745 "to an external URL."
746 )
747 super().__init__(
748 content_type=content_type,
749 status_code=status_code,
750 reason=reason,
751 charset=charset,
752 headers=headers,
753 )
754 if not 300 <= self.status_code <= 399:
755 raise ValueError(
756 "RedirectResponse status_code must be a 3xx redirect status, "
757 f"got {self.status_code}."
758 )
759 self.headers["Location"] = iri_to_uri(redirect_to) or ""
760
761 @property
762 def url(self) -> str:
763 return self.headers["Location"]
764
765 def __repr__(self) -> str:
766 return (
767 '<%(cls)s status_code=%(status_code)d%(content_type)s, url="%(url)s">' # noqa: UP031
768 % {
769 "cls": self.__class__.__name__,
770 "status_code": self.status_code,
771 "content_type": self._content_type_for_repr,
772 "url": self.url,
773 }
774 )
775
776
777class NotModifiedResponse(Response):
778 """HTTP 304 response — headers only, no Content-Type (the base class
779 skips the default for bodiless statuses).
780
781 The constructor is pinned: no content/content_type/status_code
782 parameters, so this class always means exactly "bodiless 304"."""
783
784 status_code = 304
785
786 def __init__(
787 self,
788 *,
789 reason: str | None = None,
790 charset: str | None = None,
791 headers: dict[str, Any] | None = None,
792 ):
793 super().__init__(
794 reason=reason,
795 charset=charset,
796 headers=headers,
797 )
798
799
800class NotAllowedResponse(Response):
801 """HTTP 405 response"""
802
803 status_code = 405
804
805 def __init__(
806 self,
807 permitted_methods: list[str],
808 *,
809 content_type: str | None = None,
810 status_code: int | None = None,
811 reason: str | None = None,
812 charset: str | None = None,
813 headers: dict[str, Any] | None = None,
814 ):
815 super().__init__(
816 content_type=content_type,
817 status_code=status_code,
818 reason=reason,
819 charset=charset,
820 headers=headers,
821 )
822 self.headers["Allow"] = ", ".join(permitted_methods)
823
824 def __repr__(self) -> str:
825 return "<%(cls)s [%(methods)s] status_code=%(status_code)d%(content_type)s>" % { # noqa: UP031
826 "cls": self.__class__.__name__,
827 "status_code": self.status_code,
828 "content_type": self._content_type_for_repr,
829 "methods": self.headers["Allow"],
830 }
831
832
833class JsonResponse(Response):
834 """An HTTP response class that consumes data to be serialized to JSON."""
835
836 def __init__(
837 self,
838 data: Any,
839 *,
840 encoder: type[json.JSONEncoder] = PlainJSONEncoder,
841 json_dumps_params: dict[str, Any] | None = None,
842 content_type: str = "application/json",
843 status_code: int | None = None,
844 reason: str | None = None,
845 charset: str | None = None,
846 headers: dict[str, Any] | None = None,
847 ):
848 if json_dumps_params is None:
849 json_dumps_params = {}
850 data = json.dumps(data, cls=encoder, **json_dumps_params)
851 super().__init__(
852 content=data,
853 content_type=content_type,
854 status_code=status_code,
855 reason=reason,
856 charset=charset,
857 headers=headers,
858 )