1from __future__ import annotations
2
3import codecs
4import copy
5import json
6import secrets
7import uuid
8from collections.abc import Iterator
9from functools import cached_property
10from io import BytesIO
11from itertools import chain
12from typing import TYPE_CHECKING, Any, Protocol, TypeVar, overload
13from urllib.parse import parse_qsl, quote, urlencode, urljoin, urlsplit
14
15if TYPE_CHECKING:
16 from plain.urls import ResolverMatch
17
18from plain.exceptions import ImproperlyConfigured
19from plain.http.cookie import parse_cookie, unsign_cookie_value
20from plain.http.multipartparser import (
21 MultiPartParser,
22)
23from plain.runtime import settings
24from plain.utils.datastructures import CaseInsensitiveMapping, MultiValueDict
25from plain.utils.encoding import iri_to_uri
26from plain.utils.http import parse_header_parameters
27
28from .exceptions import (
29 BadRequestError400,
30 RequestDataTooBigError400,
31 TooManyFieldsSentError400,
32 UnsupportedMediaTypeError415,
33)
34
35_T = TypeVar("_T")
36
37
38class RequestStream(Protocol):
39 """A bytes reader bound to ``Request._stream`` by the request creator
40 (server, test client, etc.). Anything providing ``read``, ``readline``,
41 and ``close`` over bytes will work.
42 """
43
44 def read(self, size: int | None = ..., /) -> bytes: ...
45 def readline(self, size: int | None = ..., /) -> bytes: ...
46 def close(self) -> None: ...
47
48
49class UnreadablePostError(OSError):
50 pass
51
52
53class RawPostDataException(Exception):
54 """
55 You cannot access raw_post_data from a request that has
56 multipart/* POST data if it has been accessed via POST,
57 FILES, etc..
58 """
59
60 pass
61
62
63class Request:
64 """A basic HTTP request."""
65
66 non_picklable_attrs = frozenset(["resolver_match", "_stream"])
67
68 # Set by the request creator (server, test client) before the view runs.
69 _stream: RequestStream
70
71 def __init__(
72 self,
73 *,
74 method: str,
75 path: str,
76 headers: dict[str, str] | None = None,
77 query_string: str = "",
78 server_scheme: str = "http",
79 server_name: str = "",
80 server_port: str = "",
81 remote_addr: str = "",
82 ):
83 self.unique_id = str(uuid.uuid4())
84 self.resolver_match: ResolverMatch | None = None
85
86 self.method = method
87 self.path = path
88 self.server_name = server_name
89 self.server_port = server_port
90 self.remote_addr = remote_addr
91 self.query_string = query_string
92 self.server_scheme = server_scheme
93 self.headers = RequestHeaders(headers or {})
94
95 # Parse content type, params, and encoding from headers
96 self.content_type: str | None
97 self.content_params: dict[str, str] | None
98 self.content_type, self.content_params = parse_header_parameters(
99 self.headers.get("Content-Type", "")
100 )
101 self.encoding: str | None = None
102 if "charset" in (self.content_params or {}):
103 try:
104 codecs.lookup(self.content_params["charset"])
105 except LookupError:
106 pass
107 else:
108 self.encoding = self.content_params["charset"]
109
110 def __repr__(self) -> str:
111 if not self.get_full_path():
112 return f"<{self.__class__.__name__}>"
113 return f"<{self.__class__.__name__}: {self.method} {self.get_full_path()!r}>"
114
115 def __getstate__(self) -> dict[str, Any]:
116 obj_dict = self.__dict__.copy()
117 for attr in self.non_picklable_attrs:
118 if attr in obj_dict:
119 del obj_dict[attr]
120 return obj_dict
121
122 def __deepcopy__(self, memo: dict[int, Any]) -> Request:
123 obj = copy.copy(self)
124 for attr in self.non_picklable_attrs:
125 if hasattr(self, attr):
126 setattr(obj, attr, copy.deepcopy(getattr(self, attr), memo))
127 memo[id(self)] = obj
128 return obj
129
130 @cached_property
131 def query_params(self) -> QueryDict:
132 return QueryDict(self.query_string, encoding=self.encoding)
133
134 @cached_property
135 def cookies(self) -> dict[str, str]:
136 return parse_cookie(self.headers.get("Cookie", ""))
137
138 @cached_property
139 def csp_nonce(self) -> str:
140 """Generate a cryptographically secure nonce for Content Security Policy.
141
142 The nonce is generated once per request and cached. It can be used in
143 CSP headers and templates to allow specific inline scripts/styles while
144 blocking others.
145 """
146 return secrets.token_urlsafe(16)
147
148 @cached_property
149 def accepted_types(self) -> list[MediaType]:
150 """Return accepted media types sorted by quality value (highest first).
151
152 When quality values are equal, the original order from the Accept header
153 is preserved (as per HTTP spec).
154 """
155 header = self.headers.get("Accept", "*/*")
156 types = [MediaType(token) for token in header.split(",") if token.strip()]
157 return sorted(types, key=lambda t: t.quality, reverse=True)
158
159 def get_preferred_type(self, *media_types: str) -> str | None:
160 """Return the most preferred media type from the given options.
161
162 Checks the Accept header in priority order (by quality value) and returns
163 the first matching media type from the provided options.
164
165 Returns None if none of the options are accepted.
166
167 Example:
168 # Accept: text/html;q=1.0, application/json;q=0.5
169 request.get_preferred_type("application/json", "text/html") # Returns "text/html"
170 """
171 for accepted in self.accepted_types:
172 for option in media_types:
173 if accepted.match(option):
174 return option
175 return None
176
177 def accepts(self, media_type: str) -> bool:
178 """Check if the given media type is accepted."""
179 return self.get_preferred_type(media_type) is not None
180
181 @cached_property
182 def host(self) -> str:
183 """
184 Return the HTTP host using the environment or request headers.
185
186 Host validation is performed by HostValidationMiddleware, so this
187 property can safely return the host without any validation.
188 """
189 # We try three options, in order of decreasing preference.
190 if settings.HTTP_X_FORWARDED_HOST and (
191 xff_host := self.headers.get("X-Forwarded-Host")
192 ):
193 host = xff_host
194 elif http_host := self.headers.get("Host"):
195 host = http_host
196 else:
197 # Reconstruct the host from server_name/port.
198 host = self.server_name
199 server_port = self.port
200 if server_port != ("443" if self.is_https() else "80"):
201 host = f"{host}:{server_port}"
202 return host
203
204 @cached_property
205 def port(self) -> str:
206 """Return the port number for the request as a string."""
207 if settings.HTTP_X_FORWARDED_PORT and (
208 xff_port := self.headers.get("X-Forwarded-Port")
209 ):
210 port = xff_port
211 else:
212 port = self.server_port
213 return str(port)
214
215 @cached_property
216 def client_ip(self) -> str:
217 """Return the client's IP address.
218
219 If HTTP_X_FORWARDED_FOR is True, checks the X-Forwarded-For header first
220 (using the first/leftmost IP). Otherwise returns REMOTE_ADDR directly.
221
222 Only enable HTTP_X_FORWARDED_FOR when behind a trusted proxy that
223 overwrites the X-Forwarded-For header.
224 """
225 if settings.HTTP_X_FORWARDED_FOR:
226 if xff := self.headers.get("X-Forwarded-For"):
227 return xff.split(",")[0].strip()
228 return self.remote_addr
229
230 @property
231 def content_length(self) -> int:
232 """Return the Content-Length header value, or 0 if not provided."""
233 try:
234 return int(self.headers.get("Content-Length") or 0)
235 except (ValueError, TypeError):
236 return 0
237
238 def get_full_path(self) -> str:
239 """Return the full path for the request, including query string."""
240 # RFC 3986 requires query string arguments to be in the ASCII range.
241 # Rather than crash if this doesn't happen, we encode defensively.
242
243 def escape_uri_path(path: str) -> str:
244 """
245 Escape the unsafe characters from the path portion of a Uniform Resource
246 Identifier (URI).
247 """
248 # These are the "reserved" and "unreserved" characters specified in RFC
249 # 3986 Sections 2.2 and 2.3:
250 # reserved = ";" | "/" | "?" | ":" | "@" | "&" | "=" | "+" | "$" | ","
251 # unreserved = alphanum | mark
252 # mark = "-" | "_" | "." | "!" | "~" | "*" | "'" | "(" | ")"
253 # The list of safe characters here is constructed subtracting ";", "=",
254 # and "?" according to RFC 3986 Section 3.3.
255 # The reason for not subtracting and escaping "/" is that we are escaping
256 # the entire path, not a path segment.
257 return quote(path, safe="/:@&+$,-_.!~*'()")
258
259 return "{}{}".format(
260 escape_uri_path(self.path),
261 ("?" + (iri_to_uri(self.query_string) or "")) if self.query_string else "",
262 )
263
264 def build_absolute_uri(self, location: str | None = None) -> str:
265 """
266 Build an absolute URI from the location and the variables available in
267 this request. If no ``location`` is specified, build the absolute URI
268 using request.get_full_path(). If the location is absolute, convert it
269 to an RFC 3987 compliant URI and return it. If location is relative or
270 is scheme-relative (i.e., ``//example.com/``), urljoin() it to a base
271 URL constructed from the request variables.
272 """
273 if location is None:
274 # Make it an absolute url (but schemeless and domainless) for the
275 # edge case that the path starts with '//'.
276 location = f"//{self.get_full_path()}"
277 else:
278 # Coerce lazy locations.
279 location = str(location)
280 bits = urlsplit(location)
281 if not (bits.scheme and bits.netloc):
282 current_scheme_host = f"{self.scheme}://{self.host}"
283
284 # Handle the simple, most common case. If the location is absolute
285 # and a scheme or host (netloc) isn't provided, skip an expensive
286 # urljoin() as long as no path segments are '.' or '..'.
287 if (
288 bits.path.startswith("/")
289 and not bits.scheme
290 and not bits.netloc
291 and "/./" not in bits.path
292 and "/../" not in bits.path
293 ):
294 # If location starts with '//' but has no netloc, reuse the
295 # schema and netloc from the current request. Strip the double
296 # slashes and continue as if it wasn't specified.
297 location = current_scheme_host + location.removeprefix("//")
298 else:
299 # Join the constructed URL with the provided location, which
300 # allows the provided location to apply query strings to the
301 # base path.
302 location = urljoin(current_scheme_host + self.path, location)
303
304 return iri_to_uri(location) or ""
305
306 @property
307 def scheme(self) -> str:
308 if settings.HTTPS_PROXY_HEADER:
309 if ":" not in settings.HTTPS_PROXY_HEADER:
310 raise ImproperlyConfigured(
311 "The HTTPS_PROXY_HEADER setting must be a string in the format "
312 "'Header-Name: value' (e.g., 'X-Forwarded-Proto: https')."
313 )
314 header, secure_value = settings.HTTPS_PROXY_HEADER.split(":", 1)
315 header = header.strip()
316 secure_value = secure_value.strip()
317 header_value = self.headers.get(header)
318 if header_value is not None:
319 header_value, *_ = header_value.split(",", 1)
320 return "https" if header_value.strip() == secure_value else "http"
321 return self.server_scheme
322
323 def is_https(self) -> bool:
324 return self.scheme == "https"
325
326 @property
327 def body(self) -> bytes:
328 if not hasattr(self, "_body"):
329 if self._read_started:
330 raise RawPostDataException(
331 "You cannot access body after reading from request's data stream"
332 )
333
334 max_size = settings.DATA_UPLOAD_MAX_MEMORY_SIZE
335
336 # Fast path: reject immediately if Content-Length exceeds the limit.
337 if max_size is not None and self.content_length > max_size:
338 raise RequestDataTooBigError400(
339 "Request body exceeded settings.DATA_UPLOAD_MAX_MEMORY_SIZE."
340 )
341
342 try:
343 if max_size is not None:
344 # Read in chunks to enforce the limit without buffering
345 # an arbitrarily large body (e.g. chunked transfers
346 # with no Content-Length).
347 chunks = []
348 bytes_read = 0
349 while True:
350 chunk = self.read(max_size - bytes_read + 1)
351 if not chunk:
352 break
353 bytes_read += len(chunk)
354 if bytes_read > max_size:
355 raise RequestDataTooBigError400(
356 "Request body exceeded settings.DATA_UPLOAD_MAX_MEMORY_SIZE."
357 )
358 chunks.append(chunk)
359 self._body = b"".join(chunks)
360 else:
361 self._body = self.read()
362 except OSError as e:
363 raise UnreadablePostError(*e.args) from e
364 finally:
365 self._stream.close()
366 self._stream = BytesIO(self._body)
367 return self._body
368
369 @cached_property
370 def _multipart_data(self) -> tuple[QueryDict, MultiValueDict]:
371 """Parse multipart/form-data. Used internally by form_data and files properties.
372
373 Raises MultiPartParserError or TooManyFilesSentError400 for malformed uploads,
374 which are handled by response_for_exception() as 400 errors.
375 """
376 return MultiPartParser(self).parse()
377
378 @cached_property
379 def json_data(self) -> dict[str, Any]:
380 """
381 Parsed JSON object from request body.
382
383 Returns dict for JSON objects.
384 Raises BadRequestError400 if JSON is invalid or not an object.
385 Raises UnsupportedMediaTypeError415 if request content-type is not JSON.
386
387 Use this when you expect JSON object data and want type-safe dict access.
388 """
389 if not self.content_type or not self.content_type.startswith(
390 "application/json"
391 ):
392 raise UnsupportedMediaTypeError415(
393 f"Request content-type is not JSON (got: {self.content_type})"
394 )
395 try:
396 parsed = json.loads(self.body)
397 except json.JSONDecodeError as e:
398 raise BadRequestError400(f"Invalid JSON in request body: {e}") from e
399
400 if not isinstance(parsed, dict):
401 raise BadRequestError400(
402 f"Expected JSON object, got {type(parsed).__name__}"
403 )
404 return parsed
405
406 @cached_property
407 def form_data(self) -> QueryDict:
408 """
409 Form data from POST body.
410
411 Returns QueryDict for application/x-www-form-urlencoded or
412 multipart/form-data content types.
413 Returns empty QueryDict if Content-Type is missing (e.g., GET requests).
414 Raises UnsupportedMediaTypeError415 if request has a different content-type with a body.
415
416 Use this when you expect form data and want type-safe QueryDict access.
417 """
418 if self.content_type == "application/x-www-form-urlencoded":
419 return QueryDict(self.body, encoding=self.encoding)
420 elif self.content_type == "multipart/form-data":
421 return self._multipart_data[0]
422 elif not self.content_type:
423 # No Content-Type (e.g., GET requests) - return empty QueryDict
424 return QueryDict(b"", encoding=self.encoding)
425 else:
426 raise UnsupportedMediaTypeError415(
427 f"Request content-type is not form data (got: {self.content_type})"
428 )
429
430 @cached_property
431 def files(self) -> MultiValueDict:
432 """
433 File uploads from multipart/form-data requests.
434
435 Returns MultiValueDict of uploaded files for multipart requests,
436 or empty MultiValueDict for other content types.
437 """
438 if self.content_type == "multipart/form-data":
439 return self._multipart_data[1]
440 return MultiValueDict()
441
442 def close(self) -> None:
443 # Close any uploaded files if they were accessed
444 if self.content_type == "multipart/form-data" and hasattr(
445 self, "_multipart_data"
446 ):
447 _, files = self._multipart_data
448 for f in chain.from_iterable(list_[1] for list_ in files.lists()):
449 f.close()
450
451 # File-like and iterator interface.
452 #
453 # Expects self._stream to be set to an appropriate source of bytes by
454 # a corresponding request creator (e.g. server or test client).
455 # Also when request data has already been read by request.json_data,
456 # request.form_data, or request.body, self._stream points to a BytesIO
457 # instance containing that data.
458
459 def read(self, *args: Any, **kwargs: Any) -> bytes:
460 self._read_started = True
461 try:
462 return self._stream.read(*args, **kwargs)
463 except OSError as e:
464 raise UnreadablePostError(*e.args) from e
465
466 def readline(self, *args: Any, **kwargs: Any) -> bytes:
467 self._read_started = True
468 try:
469 return self._stream.readline(*args, **kwargs)
470 except OSError as e:
471 raise UnreadablePostError(*e.args) from e
472
473 def __iter__(self) -> Iterator[bytes]:
474 return iter(self.readline, b"")
475
476 def readlines(self) -> list[bytes]:
477 return list(self)
478
479 def get_signed_cookie(
480 self,
481 key: str,
482 default: str | None = None,
483 salt: str = "",
484 max_age: int | None = None,
485 ) -> str | None:
486 """
487 Retrieve a cookie value signed with the SECRET_KEY.
488
489 Return default if the cookie doesn't exist or signature verification fails.
490 """
491
492 try:
493 cookie_value = self.cookies[key]
494 except KeyError:
495 return default
496
497 return unsign_cookie_value(key, cookie_value, salt, max_age, default)
498
499
500class RequestHeaders(CaseInsensitiveMapping):
501 def __init__(self, headers: dict[str, str]):
502 normalized = {
503 name.replace("_", "-").title(): value for name, value in headers.items()
504 }
505 super().__init__(normalized)
506
507 def __getitem__(self, key: str) -> str:
508 """Allow header lookup using underscores in place of hyphens."""
509 return super().__getitem__(key.replace("_", "-"))
510
511
512class QueryDict(MultiValueDict):
513 """
514 A specialized MultiValueDict which represents a query string.
515
516 A QueryDict can be used to represent GET or POST data. It subclasses
517 MultiValueDict since keys in such data can be repeated, for instance
518 in the data from a form with a <select multiple> field.
519
520 By default QueryDicts are immutable, though the copy() method
521 will always return a mutable copy.
522
523 Both keys and values set on this class are converted from the given encoding
524 (UTF-8 by default) to str.
525 """
526
527 # These are both reset in __init__, but is specified here at the class
528 # level so that unpickling will have valid values
529 _mutable = True
530 _encoding = None
531
532 def __init__(
533 self,
534 query_string: str | bytes | None = None,
535 mutable: bool = False,
536 encoding: str | None = None,
537 ):
538 super().__init__()
539 self.encoding = encoding or "utf-8"
540 query_string = query_string or ""
541 parse_qsl_kwargs: dict[str, Any] = {
542 "keep_blank_values": True,
543 "encoding": self.encoding,
544 "max_num_fields": settings.DATA_UPLOAD_MAX_NUMBER_FIELDS,
545 }
546 if isinstance(query_string, bytes):
547 # query_string normally contains URL-encoded data, a subset of ASCII.
548 query_bytes = query_string
549 try:
550 query_string = query_bytes.decode(self.encoding)
551 except UnicodeDecodeError:
552 # ... but some user agents are misbehaving :-(
553 query_string = query_bytes.decode("iso-8859-1")
554 try:
555 for key, value in parse_qsl(query_string, **parse_qsl_kwargs):
556 self.appendlist(key, value)
557 except ValueError as e:
558 # ValueError can also be raised if the strict_parsing argument to
559 # parse_qsl() is True. As that is not used by Plain, assume that
560 # the exception was raised by exceeding the value of max_num_fields
561 # instead of fragile checks of exception message strings.
562 raise TooManyFieldsSentError400(
563 "The number of GET/POST parameters exceeded "
564 "settings.DATA_UPLOAD_MAX_NUMBER_FIELDS."
565 ) from e
566 self._mutable = mutable
567
568 @classmethod
569 def fromkeys( # ty: ignore[invalid-method-override]
570 cls,
571 iterable: Any,
572 value: str = "",
573 mutable: bool = False,
574 encoding: str | None = None,
575 ) -> QueryDict:
576 """
577 Return a new QueryDict with keys (may be repeated) from an iterable and
578 values from value.
579 """
580 q = cls("", mutable=True, encoding=encoding)
581 for key in iterable:
582 q.appendlist(key, value)
583 if not mutable:
584 q._mutable = False
585 return q
586
587 @property
588 def encoding(self) -> str:
589 if self._encoding is None:
590 self._encoding = "utf-8"
591 return self._encoding
592
593 @encoding.setter
594 def encoding(self, value: str) -> None:
595 self._encoding = value
596
597 def _assert_mutable(self) -> None:
598 if not self._mutable:
599 raise AttributeError("This QueryDict instance is immutable")
600
601 def __setitem__(self, key: str, value: Any) -> None:
602 self._assert_mutable()
603 key = self.bytes_to_text(key, self.encoding)
604 value = self.bytes_to_text(value, self.encoding)
605 super().__setitem__(key, value)
606
607 def __delitem__(self, key: str) -> None:
608 self._assert_mutable()
609 super().__delitem__(key)
610
611 def __getitem__(self, key: str) -> str: # ty: ignore[invalid-method-override]
612 """
613 Return the last data value for this key as a string.
614 QueryDict values are always strings.
615 """
616 return super().__getitem__(key)
617
618 def __copy__(self) -> QueryDict:
619 result = self.__class__("", mutable=True, encoding=self.encoding)
620 for key, value in self.lists():
621 result.setlist(key, value)
622 return result
623
624 def __deepcopy__(self, memo: dict[int, Any]) -> QueryDict:
625 result = self.__class__("", mutable=True, encoding=self.encoding)
626 memo[id(self)] = result
627 for key, value in self.lists():
628 result.setlist(copy.deepcopy(key, memo), copy.deepcopy(value, memo))
629 return result
630
631 def setlist(self, key: str, list_: list[Any]) -> None:
632 self._assert_mutable()
633 key = self.bytes_to_text(key, self.encoding)
634 list_ = [self.bytes_to_text(elt, self.encoding) for elt in list_]
635 super().setlist(key, list_)
636
637 def setlistdefault(
638 self, key: str, default_list: list[Any] | None = None
639 ) -> list[Any]:
640 self._assert_mutable()
641 return super().setlistdefault(key, default_list)
642
643 def appendlist(self, key: str, value: Any) -> None:
644 self._assert_mutable()
645 key = self.bytes_to_text(key, self.encoding)
646 value = self.bytes_to_text(value, self.encoding)
647 super().appendlist(key, value)
648
649 def getlist(self, key: str, default: list[str] | None = None) -> list[str]:
650 """
651 Return the list of values for the key as strings.
652 QueryDict values are always strings.
653 """
654 return super().getlist(key, default)
655
656 @overload
657 def get(self, key: str) -> str | None: ...
658
659 @overload
660 def get(self, key: str, default: str) -> str: ...
661
662 @overload
663 def get(self, key: str, default: _T) -> str | _T: ...
664
665 def get(self, key: str, default: Any = None) -> str | Any: # ty: ignore[invalid-method-override]
666 """
667 Return the last data value for the passed key. If key doesn't exist
668 or value is an empty list, return `default`.
669
670 QueryDict values are always strings (from URL parsing), but the
671 return type preserves the type of the default parameter for type safety.
672
673 Examples:
674 get("page") -> str | None
675 get("page", "1") -> str
676 get("page", 1) -> str | int
677 """
678 return super().get(key, default)
679
680 @overload
681 def pop(self, key: str) -> str: ...
682
683 @overload
684 def pop(self, key: str, default: str) -> str: ...
685
686 @overload
687 def pop(self, key: str, default: _T) -> str | _T: ...
688
689 def pop(self, key: str, *args: Any) -> str | Any: # ty: ignore[invalid-method-override]
690 """
691 Remove and return a value for the key.
692
693 QueryDict values are always strings, but the return type preserves
694 the type of the default parameter for type safety.
695
696 Examples:
697 pop("page") -> str (or raises KeyError)
698 pop("page", "1") -> str
699 pop("page", 1) -> str | int
700 """
701 self._assert_mutable()
702 return super().pop(key, *args)
703
704 def popitem(self) -> tuple[str, Any]:
705 self._assert_mutable()
706 return super().popitem()
707
708 def clear(self) -> None:
709 self._assert_mutable()
710 super().clear()
711
712 def setdefault(self, key: str, default: Any = None) -> Any:
713 self._assert_mutable()
714 key = self.bytes_to_text(key, self.encoding)
715 default = self.bytes_to_text(default, self.encoding)
716 return super().setdefault(key, default)
717
718 def copy(self) -> QueryDict:
719 """Return a mutable copy of this object."""
720 return self.__deepcopy__({})
721
722 def urlencode(self, safe: str | None = None) -> str:
723 """
724 Return an encoded string of all query string arguments.
725
726 `safe` specifies characters which don't require quoting, for example::
727
728 >>> q = QueryDict(mutable=True)
729 >>> q['next'] = '/a&b/'
730 >>> q.urlencode()
731 'next=%2Fa%26b%2F'
732 >>> q.urlencode(safe='/')
733 'next=/a%26b/'
734 """
735 output = []
736 if safe:
737 safe_bytes: bytes = safe.encode(self.encoding)
738
739 def encode(k: bytes, v: bytes) -> str:
740 return f"{quote(k, safe_bytes)}={quote(v, safe_bytes)}"
741
742 else:
743
744 def encode(k: bytes, v: bytes) -> str:
745 return urlencode({k: v})
746
747 for k, list_ in self.lists():
748 output.extend(
749 encode(k.encode(self.encoding), str(v).encode(self.encoding))
750 for v in list_
751 )
752 return "&".join(output)
753
754 # It's neither necessary nor appropriate to use
755 # plain.utils.encoding.force_str() for parsing URLs and form inputs. Thus,
756 # this slightly more restricted function, used by QueryDict.
757 @staticmethod
758 def bytes_to_text(s: Any, encoding: str) -> str:
759 """
760 Convert bytes objects to strings, using the given encoding. Illegally
761 encoded input characters are replaced with Unicode "unknown" codepoint
762 (\ufffd).
763
764 Return any non-bytes objects without change.
765 """
766 if isinstance(s, bytes):
767 return str(s, encoding, "replace")
768 else:
769 return s
770
771
772class MediaType:
773 def __init__(self, media_type_raw_line: str | MediaType):
774 line = str(media_type_raw_line) if media_type_raw_line else ""
775 full_type, self.params = parse_header_parameters(line)
776 self.main_type, _, self.sub_type = full_type.partition("/")
777
778 def __str__(self) -> str:
779 params_str = "".join(f"; {k}={v}" for k, v in self.params.items())
780 return "{}{}{}".format(
781 self.main_type,
782 (f"/{self.sub_type}") if self.sub_type else "",
783 params_str,
784 )
785
786 def __repr__(self) -> str:
787 return f"<{self.__class__.__qualname__}: {self}>"
788
789 @property
790 def is_all_types(self) -> bool:
791 return self.main_type == "*" and self.sub_type == "*"
792
793 @property
794 def quality(self) -> float:
795 """Return the quality value from the Accept header (default 1.0)."""
796 return float(self.params.get("q", 1.0))
797
798 def match(self, other: str | MediaType) -> bool:
799 if self.is_all_types:
800 return True
801 other = MediaType(other)
802 if self.main_type == other.main_type and self.sub_type in {"*", other.sub_type}:
803 return True
804 return False