Merge branch 'main' into accept-enhance

sanic/headers.py

@@ -35,141 +35,96 @@ _host_re = re.compile(
 
 def parse_arg_as_accept(f):
     def func(self, other, *args, **kwargs):
-        if not isinstance(other, Accept) and other:
+        if not isinstance(other, MediaType) and other:
-            other = Accept.parse(other)
+            other = MediaType._parse(other)
         return f(self, other, *args, **kwargs)
 
     return func
 
 
-class MediaType(str):
+class MediaType:
-    def __new__(cls, value: str):
+    """A media type, as used in the Accept header."""
-        return str.__new__(cls, value)
-
-    def __init__(self, value: str) -> None:
-        self.value = value
-        self.is_wildcard = self.check_if_wildcard(value)
-
-    def __eq__(self, other):
-        if self.is_wildcard:
-            return True
-
-        if self.match(other):
-            return True
-
-        other_is_wildcard = (
-            other.is_wildcard
-            if isinstance(other, MediaType)
-            else self.check_if_wildcard(other)
-        )
-
-        return other_is_wildcard
-
-    def match(self, other):
-        other_value = other.value if isinstance(other, MediaType) else other
-        return self.value == other_value
-
-    @staticmethod
-    def check_if_wildcard(value):
-        return value == "*"
-
-
-class Accept(str):
-    def __new__(cls, value: str, *args, **kwargs):
-        return str.__new__(cls, value)
 
     def __init__(
         self,
-        value: str,
+        type_: str,
-        type_: MediaType,
+        subtype: str,
-        subtype: MediaType,
+        **params: str,
-        *,
-        q: str = "1.0",
-        **kwargs: str,
     ):
-        qvalue = float(q)
-        if qvalue > 1 or qvalue < 0:
-            raise InvalidHeader(
-                f"Accept header qvalue must be between 0 and 1, not: {qvalue}"
-            )
-        self.value = value
         self.type_ = type_
         self.subtype = subtype
-        self.qvalue = qvalue
+        self.q = float(params.get("q", "1.0"))
-        self.params = kwargs
+        self.params = params
+        self.mime = f"{type_}/{subtype}"
 
-    def _compare(self, other, method):
+    def __repr__(self):
-        try:
+        return self.mime + "".join(f";{k}={v}" for k, v in self.params.items())
-            return method(self.qvalue, other.qvalue)
-        except (AttributeError, TypeError):
-            return NotImplemented
 
-    @parse_arg_as_accept
+    def __eq__(self, other):
-    def __lt__(self, other: Union[str, Accept]):
+        """Check for mime (str or MediaType) identical type/subtype."""
-        return self._compare(other, lambda s, o: s < o)
+        if isinstance(other, str):
+            return self.mime == other
+        if isinstance(other, MediaType):
+            return self.mime == other.mime
+        return NotImplemented
 
-    @parse_arg_as_accept
-    def __le__(self, other: Union[str, Accept]):
-        return self._compare(other, lambda s, o: s <= o)
-
-    @parse_arg_as_accept
-    def __eq__(self, other: Union[str, Accept]):  # type: ignore
-        return self._compare(other, lambda s, o: s == o)
-
-    @parse_arg_as_accept
-    def __ge__(self, other: Union[str, Accept]):
-        return self._compare(other, lambda s, o: s >= o)
-
-    @parse_arg_as_accept
-    def __gt__(self, other: Union[str, Accept]):
-        return self._compare(other, lambda s, o: s > o)
-
-    @parse_arg_as_accept
-    def __ne__(self, other: Union[str, Accept]):  # type: ignore
-        return self._compare(other, lambda s, o: s != o)
-
-    @parse_arg_as_accept
     def match(
         self,
-        other,
+        mime: str,
-        *,
+        allow_type_wildcard=True,
-        allow_type_wildcard: bool = True,
+        allow_subtype_wildcard=True,
-        allow_subtype_wildcard: bool = True,
+    ) -> Optional[MediaType]:
-    ) -> bool:
+        """Check if this media type matches the given mime type/subtype.
-        type_match = (
+
-            self.type_ == other.type_
+        Wildcards are supported both ways on both type and subtype.
-            if allow_type_wildcard
+
-            else (
+        Note:  Use the `==` operator instead to check for literal matches
-                self.type_.match(other.type_)
+        without expanding wildcards.
-                and not self.type_.is_wildcard
+
-                and not other.type_.is_wildcard
+        @param media_type: A type/subtype string to match.
-            )
+        @return `self` if the media types are compatible, else `None`
-        )
+        """
-        subtype_match = (
+        mt = MediaType._parse(mime)
-            self.subtype == other.subtype
+        return (
-            if allow_subtype_wildcard
+            self
-            else (
+            if (
-                self.subtype.match(other.subtype)
+                # Subtype match
-                and not self.subtype.is_wildcard
+                (self.subtype in (mt.subtype, "*") or mt.subtype == "*")
-                and not other.subtype.is_wildcard
+                # Type match
+                and (self.type_ in (mt.type_, "*") or mt.type_ == "*")
+                # Allow disabling wildcards (backwards compatibility with tests)
+                and (
+                    allow_type_wildcard
+                    or self.type_ != "*"
+                    and mt.type_ != "*"
+                )
+                and (
+                    allow_subtype_wildcard
+                    or self.subtype != "*"
+                    and mt.subtype != "*"
+                )
             )
+            else None
         )
 
-        return type_match and subtype_match
+    @property
+    def has_wildcard(self) -> bool:
+        """Return True if this media type has a wildcard in it."""
+        return "*" in (self.subtype, self.type_)
+
+    @property
+    def is_wildcard(self) -> bool:
+        """Return True if this is the wildcard `*/*`"""
+        return self.type_ == "*" and self.subtype == "*"
 
     @classmethod
-    def parse(cls, raw: str) -> Accept:
+    def _parse(cls, mime_with_params: str) -> MediaType:
-        invalid = False
+        mtype = mime_with_params.strip()
-        mtype = raw.strip()
 
-        try:
+        media, *raw_params = mtype.split(";")
-            media, *raw_params = mtype.split(";")
+        type_, subtype = media.split("/", 1)
-            type_, subtype = media.split("/")
+        if not type_ or not subtype:
-        except ValueError:
+            raise ValueError(f"Invalid media type: {mtype}")
-            invalid = True
-
-        if invalid or not type_ or not subtype:
-            raise InvalidHeader(f"Header contains invalid Accept value: {raw}")
 
         params = dict(
             [
@@ -178,28 +133,79 @@ class Accept(str):
             ]
         )
 
-        return cls(mtype, MediaType(type_), MediaType(subtype), **params)
+        return cls(type_.lstrip(), subtype.rstrip(), **params)
 
 
-class AcceptContainer(list):
+class Matched(str):
-    def __contains__(self, o: object) -> bool:
+    """A matching result of a MIME string against a MediaType."""
-        return any(item.match(o) for item in self)
 
-    def match(
+    def __new__(cls, mime: str, m: Optional[MediaType]):
-        self,
+        return super().__new__(cls, mime)
-        o: object,
+
-        *,
+    def __init__(self, mime: str, m: Optional[MediaType]):
-        allow_type_wildcard: bool = True,
+        self.m = m
-        allow_subtype_wildcard: bool = True,
+
-    ) -> bool:
+    def __repr__(self):
-        return any(
+        return f"<{self} matched {self.m}>" if self else "<no match>"
-            item.match(
+
-                o,
+
-                allow_type_wildcard=allow_type_wildcard,
+class AcceptList(list):
-                allow_subtype_wildcard=allow_subtype_wildcard,
+    """A list of media types, as used in the Accept header.
-            )
+
-            for item in self
+    The Accept header entries are listed in order of preference, starting
+    with the most preferred. This class is a list of `MediaType` objects,
+    that encapsulate also the q value or any other parameters.
+
+    Two separate methods are provided for searching the list:
+    - 'match' for finding the most preferred match (wildcards supported)
+    -  operator 'in' for checking explicit matches (wildcards as literals)
+    """
+
+    def match(self, *mimes: str) -> Matched:
+        """Find a media type accepted by the client.
+
+        This method can be used to find which of the media types requested by
+        the client is most preferred against the ones given as arguments.
+
+        The ordering of preference is set by:
+        1. The q values on the Accept header, and those being equal,
+        2. The order of the arguments (first is most preferred), and
+        3. The first matching entry on the Accept header.
+
+        Wildcards are matched both ways. A match is usually found, as the
+        Accept headers typically include `*/*`, in particular if the header
+        is missing, is not manually set, or if the client is a browser.
+
+        Note: the returned object behaves as a string of the mime argument
+        that matched, and is empty/falsy if no match was found. The matched
+        header entry `MediaType` or `None` is available as the `m` attribute.
+
+        @param mimes: Any MIME types to search for in order of preference.
+        @return A match object with the mime string and the MediaType object.
+        """
+        l = sorted(
+            [
+                (-acc.q, i, j, mime, acc)  # Sort by -q, i, j
+                for j, acc in enumerate(self)
+                for i, mime in enumerate(mimes)
+                if acc.match(mime)
+            ]
         )
+        return Matched(*(l[0][3:] if l else ("", None)))
+
+
+def parse_accept(accept: str) -> AcceptList:
+    """Parse an Accept header and order the acceptable media types in
+    accorsing to RFC 7231, s. 5.3.2
+    https://datatracker.ietf.org/doc/html/rfc7231#section-5.3.2
+    """
+    if not accept:
+        return AcceptList()
+    try:
+        a = [MediaType._parse(mtype) for mtype in accept.split(",")]
+        return AcceptList(sorted(a, key=lambda mtype: -mtype.q))
+    except ValueError:
+        raise InvalidHeader(f"Invalid header value in Accept: {accept}")
 
 
 def parse_content_header(value: str) -> Tuple[str, Options]:
@@ -368,34 +374,6 @@ def format_http1_response(status: int, headers: HeaderBytesIterable) -> bytes:
     return ret
 
 
-def _sort_accept_value(accept: Accept):
-    return (
-        accept.qvalue,
-        len(accept.params),
-        accept.subtype != "*",
-        accept.type_ != "*",
-    )
-
-
-def parse_accept(accept: str) -> AcceptContainer:
-    """Parse an Accept header and order the acceptable media types in
-    accorsing to RFC 7231, s. 5.3.2
-    https://datatracker.ietf.org/doc/html/rfc7231#section-5.3.2
-    """
-    media_types = accept.split(",")
-    accept_list: List[Accept] = []
-
-    for mtype in media_types:
-        if not mtype:
-            continue
-
-        accept_list.append(Accept.parse(mtype))
-
-    return AcceptContainer(
-        sorted(accept_list, key=_sort_accept_value, reverse=True)
-    )
-
-
 def parse_credentials(
     header: Optional[str],
     prefixes: Union[List, Tuple, Set] = None,

sanic/request.py

@@ -47,7 +47,7 @@ from sanic.constants import (
 )
 from sanic.exceptions import BadRequest, BadURL, ServerError
 from sanic.headers import (
-    AcceptContainer,
+    AcceptList,
     Options,
     parse_accept,
     parse_content_header,
@@ -167,7 +167,7 @@ class Request:
         self.conn_info: Optional[ConnInfo] = None
         self.ctx = SimpleNamespace()
         self.parsed_forwarded: Optional[Options] = None
-        self.parsed_accept: Optional[AcceptContainer] = None
+        self.parsed_accept: Optional[AcceptList] = None
         self.parsed_credentials: Optional[Credentials] = None
         self.parsed_json = None
         self.parsed_form: Optional[RequestParameters] = None
@@ -499,7 +499,7 @@ class Request:
         return self.parsed_json
 
     @property
-    def accept(self) -> AcceptContainer:
+    def accept(self) -> AcceptList:
         """
         :return: The ``Accept`` header parsed
         :rtype: AcceptContainer

tests/test_headers.py

@@ -185,30 +185,22 @@ def test_request_line(app):
 
     assert request.request_line == b"GET / HTTP/1.1"
 
-
 @pytest.mark.parametrize(
     "raw",
     (
-        "show/first, show/second",
+        "text/html, application/xhtml+xml, application/xml;q=0.9, */*;q=0.8",
-        "show/*, show/first",
+        "application/xml;q=0.9, */*;q=0.8, text/html, application/xhtml+xml",
-        "*/*, show/first",
+        "foo/bar;q=0.9, */*;q=0.8, text/html=0.8, text/plain, application/xhtml+xml",
-        "*/*, show/*",
-        "other/*; q=0.1, show/*; q=0.2",
-        "show/first; q=0.5, show/second; q=0.5",
-        "show/first; foo=bar, show/second; foo=bar",
-        "show/second, show/first; foo=bar",
-        "show/second; q=0.5, show/first; foo=bar; q=0.5",
-        "show/second; q=0.5, show/first; q=1.0",
-        "show/first, show/second; q=1.0",
-    ),
-)
-def test_parse_accept_ordered_okay(raw):
-    ordered = headers.parse_accept(raw)
-    expected_subtype = (
-        "*" if all(q.subtype.is_wildcard for q in ordered) else "first"
     )
-    assert ordered[0].type_ == "show"
+)
-    assert ordered[0].subtype == expected_subtype
+def test_accept_ordering(raw):
+    """Should sort by q but also be stable."""
+    accept = headers.parse_accept(raw)
+    assert accept[0].type_ == "text"
+    raw1 =  ", ".join(str(a) for a in accept)
+    accept = headers.parse_accept(raw1)
+    raw2 =  ", ".join(str(a) for a in accept)
+    assert raw1 == raw2
 
 
 @pytest.mark.parametrize(
@@ -225,40 +217,27 @@ def test_bad_accept(raw):
 
 
 def test_empty_accept():
-    assert headers.parse_accept("") == []
+    a = headers.parse_accept("")
+    assert a == []
+    assert not a.match("*/*")
 
 
 def test_wildcard_accept_set_ok():
     accept = headers.parse_accept("*/*")[0]
-    assert accept.type_.is_wildcard
+    assert accept.is_wildcard
-    assert accept.subtype.is_wildcard
+    assert accept.has_wildcard
+
+    accept = headers.parse_accept("foo/*")[0]
+    assert not accept.is_wildcard
+    assert accept.has_wildcard
+
+    accept = headers.parse_accept("*/bar")[0]
+    assert not accept.is_wildcard
+    assert accept.has_wildcard
 
     accept = headers.parse_accept("foo/bar")[0]
-    assert not accept.type_.is_wildcard
+    assert not accept.is_wildcard
-    assert not accept.subtype.is_wildcard
+    assert not accept.has_wildcard
-
-
-def test_accept_parsed_against_str():
-    accept = headers.Accept.parse("foo/bar")
-    assert accept > "foo/bar; q=0.1"
-
-
-def test_media_type_equality():
-    assert headers.MediaType("foo") == headers.MediaType("foo") == "foo"
-    assert headers.MediaType("foo") == headers.MediaType("*") == "*"
-    assert headers.MediaType("foo") != headers.MediaType("bar")
-    assert headers.MediaType("foo") != "bar"
-
-
-def test_media_type_matching():
-    assert headers.MediaType("foo").match(headers.MediaType("foo"))
-    assert headers.MediaType("foo").match("foo")
-
-    assert not headers.MediaType("foo").match(headers.MediaType("*"))
-    assert not headers.MediaType("foo").match("*")
-
-    assert not headers.MediaType("foo").match(headers.MediaType("bar"))
-    assert not headers.MediaType("foo").match("bar")
 
 
 @pytest.mark.parametrize(
@@ -266,87 +245,52 @@ def test_media_type_matching():
     (
         # ALLOW BOTH
         ("foo/bar", "foo/bar", True, True, True),
-        ("foo/bar", headers.Accept.parse("foo/bar"), True, True, True),
         ("foo/bar", "foo/*", True, True, True),
-        ("foo/bar", headers.Accept.parse("foo/*"), True, True, True),
         ("foo/bar", "*/*", True, True, True),
-        ("foo/bar", headers.Accept.parse("*/*"), True, True, True),
         ("foo/*", "foo/bar", True, True, True),
-        ("foo/*", headers.Accept.parse("foo/bar"), True, True, True),
         ("foo/*", "foo/*", True, True, True),
-        ("foo/*", headers.Accept.parse("foo/*"), True, True, True),
         ("foo/*", "*/*", True, True, True),
-        ("foo/*", headers.Accept.parse("*/*"), True, True, True),
         ("*/*", "foo/bar", True, True, True),
-        ("*/*", headers.Accept.parse("foo/bar"), True, True, True),
         ("*/*", "foo/*", True, True, True),
-        ("*/*", headers.Accept.parse("foo/*"), True, True, True),
         ("*/*", "*/*", True, True, True),
-        ("*/*", headers.Accept.parse("*/*"), True, True, True),
         # ALLOW TYPE
         ("foo/bar", "foo/bar", True, True, False),
-        ("foo/bar", headers.Accept.parse("foo/bar"), True, True, False),
         ("foo/bar", "foo/*", False, True, False),
-        ("foo/bar", headers.Accept.parse("foo/*"), False, True, False),
         ("foo/bar", "*/*", False, True, False),
-        ("foo/bar", headers.Accept.parse("*/*"), False, True, False),
         ("foo/*", "foo/bar", False, True, False),
-        ("foo/*", headers.Accept.parse("foo/bar"), False, True, False),
         ("foo/*", "foo/*", False, True, False),
-        ("foo/*", headers.Accept.parse("foo/*"), False, True, False),
         ("foo/*", "*/*", False, True, False),
-        ("foo/*", headers.Accept.parse("*/*"), False, True, False),
         ("*/*", "foo/bar", False, True, False),
-        ("*/*", headers.Accept.parse("foo/bar"), False, True, False),
         ("*/*", "foo/*", False, True, False),
-        ("*/*", headers.Accept.parse("foo/*"), False, True, False),
         ("*/*", "*/*", False, True, False),
-        ("*/*", headers.Accept.parse("*/*"), False, True, False),
         # ALLOW SUBTYPE
         ("foo/bar", "foo/bar", True, False, True),
-        ("foo/bar", headers.Accept.parse("foo/bar"), True, False, True),
         ("foo/bar", "foo/*", True, False, True),
-        ("foo/bar", headers.Accept.parse("foo/*"), True, False, True),
         ("foo/bar", "*/*", False, False, True),
-        ("foo/bar", headers.Accept.parse("*/*"), False, False, True),
         ("foo/*", "foo/bar", True, False, True),
-        ("foo/*", headers.Accept.parse("foo/bar"), True, False, True),
         ("foo/*", "foo/*", True, False, True),
-        ("foo/*", headers.Accept.parse("foo/*"), True, False, True),
         ("foo/*", "*/*", False, False, True),
-        ("foo/*", headers.Accept.parse("*/*"), False, False, True),
         ("*/*", "foo/bar", False, False, True),
-        ("*/*", headers.Accept.parse("foo/bar"), False, False, True),
         ("*/*", "foo/*", False, False, True),
-        ("*/*", headers.Accept.parse("foo/*"), False, False, True),
         ("*/*", "*/*", False, False, True),
-        ("*/*", headers.Accept.parse("*/*"), False, False, True),
     ),
 )
 def test_accept_matching(value, other, outcome, allow_type, allow_subtype):
     assert (
-        headers.Accept.parse(value).match(
+        bool(headers.MediaType._parse(value).match(
             other,
             allow_type_wildcard=allow_type,
             allow_subtype_wildcard=allow_subtype,
-        )
+        ))
         is outcome
     )
 
 
-@pytest.mark.parametrize("value", ("foo/bar", "foo/*", "*/*"))
-def test_value_in_accept(value):
-    acceptable = headers.parse_accept(value)
-    assert "foo/bar" in acceptable
-    assert "foo/*" in acceptable
-    assert "*/*" in acceptable
-
-
 @pytest.mark.parametrize("value", ("foo/bar", "foo/*"))
 def test_value_not_in_accept(value):
     acceptable = headers.parse_accept(value)
-    assert "no/match" not in acceptable
+    assert "*/*" not in acceptable
-    assert "no/*" not in acceptable
+    assert "*/bar" not in acceptable
 
 
 @pytest.mark.parametrize(
@@ -355,16 +299,22 @@ def test_value_not_in_accept(value):
         (
             "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",  # noqa: E501
             [
-                "text/html",
+                ("text/html", 1.0),
-                "application/xhtml+xml",
+                ("application/xhtml+xml", 1.0),
-                "image/avif",
+                ("image/avif", 1.0),
-                "image/webp",
+                ("image/webp", 1.0),
-                "application/xml;q=0.9",
+                ("application/xml", 0.9),
-                "*/*;q=0.8",
+                ("*/*", 0.8),
             ],
         ),
     ),
 )
 def test_browser_headers(header, expected):
+    mimes = [e[0] for e in expected]
+    qs = [e[1] for e in expected]
     request = Request(b"/", {"accept": header}, "1.1", "GET", None, None)
-    assert request.accept == expected
+    assert request.accept == mimes
+    for a, m, q in zip(request.accept, mimes, qs):
+        assert a == m
+        assert a.str == m
+        assert a.q == q