Add response.is_json property

aio_request/base.py

@@ -1,5 +1,6 @@
 import abc
 import json
+import re
 from typing import Any, Callable, Dict, Iterable, List, Mapping, Optional, Tuple, Union
 
 import multidict
@@ -29,6 +30,9 @@ class Header:
     Mapping[Union[str, multidict.istr], str], multidict.CIMultiDictProxy[str], multidict.CIMultiDict[str]
 ]
 
+
+json_re = re.compile(r"^application/(?:[\w.+-]+?\+)?json", re.RegexFlag.IGNORECASE)
+
 PathParameters = Mapping[str, Any]
 QueryParameters = Union[Mapping[str, Any], Iterable[Tuple[str, Any]], _MultiDict]
 Headers = _MultiDict
@@ -125,6 +129,14 @@ def is_client_error(self) -> bool:
     def is_server_error(self) -> bool:
         return 500 <= self.status < 600
 
+    @property
+    def content_type(self) -> Optional[str]:
+        return self.headers.get(Header.CONTENT_TYPE)
+
+    @property
+    def is_json(self) -> bool:
+        return bool(json_re.match(self.content_type or ""))
+
 
 class ClosableResponse(Response, Closable):
     __slots__ = ()
@@ -135,18 +147,19 @@ async def close(self) -> None:
 
 
 class EmptyResponse(ClosableResponse):
-    __slots__ = ("_status",)
+    __slots__ = ("_status", "_headers")
 
-    def __init__(self, *, status: int):
+    def __init__(self, *, status: int, headers: multidict.CIMultiDictProxy[str] = EMPTY_HEADERS):
         self._status = status
+        self._headers = headers
 
     @property
     def status(self) -> int:
         return self._status
 
     @property
     def headers(self) -> multidict.CIMultiDictProxy[str]:
-        return EMPTY_HEADERS
+        return self._headers
 
     async def json(
         self,

tests/test_base.py

@@ -1,3 +1,6 @@
+import multidict
+import pytest
+
 import aio_request
 
 
@@ -6,3 +9,21 @@ async def test_update_headers():
     request = request.update_headers({"c": "d"})
 
     assert request.headers == {"a": "b", "c": "d"}
+
+
+@pytest.mark.parametrize(
+    "is_json, response_content_type, content_type",
+    [
+        (False, "", "application/json"),
+        (False, "application/xml", "application/json"),
+        (True, "application/json", "application/json"),
+        (True, "application/problem+json", "application/json"),
+    ],
+)
+async def test_response_is_json(is_json: bool, response_content_type: str, content_type: str):
+    headers = multidict.CIMultiDict[str]()
+    headers.add(aio_request.Header.CONTENT_TYPE, response_content_type)
+    response = aio_request.EmptyResponse(
+        status=200, headers=multidict.CIMultiDictProxy[str](headers),
+    )
+    assert is_json == response.is_json