Replace request mimetype with content_type

Author: p1c2u

openapi_core/contrib/aiohttp/requests.py

@@ -49,5 +49,5 @@ def body(self) -> str | None:
         return self._body
 
     @property
-    def mimetype(self) -> str:
+    def content_type(self) -> str:
         return self.request.content_type

openapi_core/contrib/django/requests.py

@@ -81,5 +81,7 @@ def body(self) -> str:
         return self.request.body.decode("utf-8")
 
     @property
-    def mimetype(self) -> str:
-        return self.request.content_type or ""
+    def content_type(self) -> str:
+        content_type = self.request.META.get("CONTENT_TYPE", "")
+        assert isinstance(content_type, str)
+        return content_type

openapi_core/contrib/falcon/requests.py

@@ -56,10 +56,10 @@ def body(self) -> Optional[str]:
         return dumps(getattr(self.request, "json", media))
 
     @property
-    def mimetype(self) -> str:
+    def content_type(self) -> str:
         if self.request.content_type:
             assert isinstance(self.request.content_type, str)
-            return self.request.content_type.partition(";")[0]
+            return self.request.content_type
 
         assert isinstance(self.request.options, RequestOptions)
         assert isinstance(self.request.options.default_media_type, str)

openapi_core/contrib/requests/requests.py

@@ -74,13 +74,13 @@ def body(self) -> Optional[str]:
         return self.request.body
 
     @property
-    def mimetype(self) -> str:
+    def content_type(self) -> str:
         # Order matters because all python requests issued from a session
         # include Accept */* which does not necessarily match the content type
         return str(
             self.request.headers.get("Content-Type")
             or self.request.headers.get("Accept")
-        ).split(";")[0]
+        )
 
 
 class RequestsOpenAPIWebhookRequest(RequestsOpenAPIRequest):

openapi_core/contrib/starlette/requests.py

@@ -44,9 +44,9 @@ def body(self) -> Optional[str]:
         return body
 
     @property
-    def mimetype(self) -> str:
-        content_type = self.request.headers.get("Content-Type")
-        if content_type:
-            return content_type.partition(";")[0]
-
-        return ""
+    def content_type(self) -> str:
+        # default value according to RFC 2616
+        return (
+            self.request.headers.get("Content-Type")
+            or "application/octet-stream"
+        )

openapi_core/contrib/werkzeug/requests.py

@@ -43,8 +43,9 @@ def body(self) -> Optional[str]:
         return self.request.get_data(as_text=True)
 
     @property
-    def mimetype(self) -> str:
-        return self.request.mimetype
+    def content_type(self) -> str:
+        # default value according to RFC 2616
+        return self.request.content_type or "application/octet-stream"
 
     def get_path(self, path: str) -> str:
         return "".join([self.request.root_path, path])

openapi_core/protocols.py

@@ -21,7 +21,7 @@ def body(self) -> Optional[str]:
         ...
 
     @property
-    def mimetype(self) -> str:
+    def content_type(self) -> str:
         ...
 
 
@@ -48,11 +48,9 @@ class Request(BaseRequest, Protocol):
             to write resolved path parameters.
         body
             The request body, as string.
-        mimetype
-            Like content type, but without parameters (eg, without charset,
-            type etc.) and always lowercase.
-            For example if the content type is "text/HTML; charset=utf-8"
-            the mimetype would be "text/html".
+        content_type
+            The content type with parameters (eg, charset, boundary etc.)
+            and always lowercase.
     """
 
     @property
@@ -78,11 +76,9 @@ class WebhookRequest(BaseRequest, Protocol):
             to write resolved path parameters.
         body
             The request body, as string.
-        mimetype
-            Like content type, but without parameters (eg, without charset,
-            type etc.) and always lowercase.
-            For example if the content type is "text/HTML; charset=utf-8"
-            the mimetype would be "text/html".
+        content_type
+            The content type with parameters (eg, charset, boundary etc.)
+            and always lowercase.
     """
 
     @property

openapi_core/testing/requests.py

@@ -21,7 +21,7 @@ def __init__(
         headers: Optional[Dict[str, Any]] = None,
         cookies: Optional[Dict[str, Any]] = None,
         data: Optional[str] = None,
-        mimetype: str = "application/json",
+        content_type: str = "application/json",
     ):
         self.host_url = host_url
         self.method = method.lower()
@@ -32,7 +32,7 @@ def __init__(
         self.headers = headers
         self.cookies = cookies
         self.body = data or ""
-        self.mimetype = mimetype
+        self.content_type = content_type
 
         self.parameters = RequestParameters(
             path=self.view_args or {},

openapi_core/unmarshalling/request/unmarshallers.py

@@ -150,7 +150,9 @@ def _unmarshal(
             params_errors = []
 
         try:
-            body = self._get_body(request.body, request.mimetype, operation)
+            body = self._get_body(
+                request.body, request.content_type, operation
+            )
         except MissingRequestBody:
             body = None
             body_errors = []
@@ -172,7 +174,9 @@ def _unmarshal_body(
         self, request: BaseRequest, operation: SchemaPath, path: SchemaPath
     ) -> RequestUnmarshalResult:
         try:
-            body = self._get_body(request.body, request.mimetype, operation)
+            body = self._get_body(
+                request.body, request.content_type, operation
+            )
         except MissingRequestBody:
             body = None
             errors = []

openapi_core/validation/request/validators.py

@@ -111,15 +111,15 @@ def _iter_errors(
             yield from exc.errors
 
         try:
-            self._get_body(request.body, request.mimetype, operation)
+            self._get_body(request.body, request.content_type, operation)
         except RequestBodyValidationError as exc:
             yield exc
 
     def _iter_body_errors(
         self, request: BaseRequest, operation: SchemaPath
     ) -> Iterator[Exception]:
         try:
-            self._get_body(request.body, request.mimetype, operation)
+            self._get_body(request.body, request.content_type, operation)
         except RequestBodyValidationError as exc:
             yield exc
 

tests/integration/test_petstore.py

@@ -1015,7 +1015,7 @@ def test_post_pets_raises_invalid_mimetype(self, spec):
             "/pets",
             path_pattern=path_pattern,
             data=data,
-            mimetype="text/html",
+            content_type="text/html",
             headers=headers,
             cookies=cookies,
         )
@@ -1150,7 +1150,7 @@ def test_post_pets_raises_invalid_server_error(self, spec):
             "/pets",
             path_pattern=path_pattern,
             data=data,
-            mimetype="text/html",
+            content_type="text/html",
             headers=headers,
             cookies=cookies,
         )

tests/integration/unmarshalling/test_request_unmarshaller.py

@@ -186,7 +186,7 @@ def test_invalid_content_type(self, request_unmarshaller):
             "post",
             "/v1/pets",
             path_pattern="/v1/pets",
-            mimetype="text/csv",
+            content_type="text/csv",
             data=data,
             headers=headers,
             cookies=cookies,
@@ -349,7 +349,7 @@ def test_post_pets_plain_no_schema(self, request_unmarshaller):
             data=data,
             headers=headers,
             cookies=cookies,
-            mimetype="text/plain",
+            content_type="text/plain",
         )
 
         result = request_unmarshaller.unmarshal(request)

tests/integration/validation/test_request_validators.py

@@ -101,7 +101,7 @@ def test_media_type_not_found(self, request_validator):
             "post",
             "/v1/pets",
             path_pattern="/v1/pets",
-            mimetype="text/csv",
+            content_type="text/csv",
             data=data,
             headers=headers,
             cookies=cookies,

tests/unit/contrib/django/test_django.py

@@ -84,7 +84,7 @@ def test_no_resolver(self, request_factory):
         assert openapi_request.path == request.path
         assert openapi_request.path_pattern is None
         assert openapi_request.body == ""
-        assert openapi_request.mimetype == request.content_type
+        assert openapi_request.content_type == request.content_type
 
     def test_simple(self, request_factory):
         from django.urls import resolve
@@ -105,7 +105,7 @@ def test_simple(self, request_factory):
         assert openapi_request.path == request.path
         assert openapi_request.path_pattern == request.path
         assert openapi_request.body == ""
-        assert openapi_request.mimetype == request.content_type
+        assert openapi_request.content_type == request.content_type
 
     def test_url_rule(self, request_factory):
         from django.urls import resolve
@@ -126,7 +126,7 @@ def test_url_rule(self, request_factory):
         assert openapi_request.path == request.path
         assert openapi_request.path_pattern == "/admin/auth/group/{object_id}/"
         assert openapi_request.body == ""
-        assert openapi_request.mimetype == request.content_type
+        assert openapi_request.content_type == request.content_type
 
     def test_url_regexp_pattern(self, request_factory):
         from django.urls import resolve
@@ -147,7 +147,7 @@ def test_url_regexp_pattern(self, request_factory):
         assert openapi_request.path == request.path
         assert openapi_request.path_pattern == request.path
         assert openapi_request.body == ""
-        assert openapi_request.mimetype == request.content_type
+        assert openapi_request.content_type == request.content_type
 
     def test_drf_default_value_pattern(self, request_factory):
         from django.urls import resolve
@@ -168,7 +168,7 @@ def test_drf_default_value_pattern(self, request_factory):
         assert openapi_request.path == request.path
         assert openapi_request.path_pattern == "/object/{pk}/action/"
         assert openapi_request.body == ""
-        assert openapi_request.mimetype == request.content_type
+        assert openapi_request.content_type == request.content_type
 
 
 class TestDjangoOpenAPIResponse(BaseTestDjango):

tests/unit/contrib/flask/test_flask_requests.py

@@ -32,7 +32,7 @@ def test_simple(self, request_factory, request):
         assert openapi_request.host_url == request.host_url
         assert openapi_request.path == request.path
         assert openapi_request.body == ""
-        assert openapi_request.mimetype == request.mimetype
+        assert openapi_request.content_type == "application/octet-stream"
 
     def test_multiple_values(self, request_factory, request):
         request = request_factory(
@@ -60,7 +60,7 @@ def test_multiple_values(self, request_factory, request):
         assert openapi_request.host_url == request.host_url
         assert openapi_request.path == request.path
         assert openapi_request.body == ""
-        assert openapi_request.mimetype == request.mimetype
+        assert openapi_request.content_type == "application/octet-stream"
 
     def test_url_rule(self, request_factory, request):
         request = request_factory("GET", "/browse/12/", subdomain="kb")
@@ -82,4 +82,4 @@ def test_url_rule(self, request_factory, request):
         assert openapi_request.path == request.path
         assert openapi_request.path_pattern == "/browse/{id}/"
         assert openapi_request.body == ""
-        assert openapi_request.mimetype == request.mimetype
+        assert openapi_request.content_type == "application/octet-stream"

tests/unit/contrib/requests/test_requests_requests.py

@@ -31,7 +31,7 @@ def test_simple(self, request_factory, request):
         assert openapi_request.host_url == "http://localhost"
         assert openapi_request.path == "/"
         assert openapi_request.body == prepared.body
-        assert openapi_request.mimetype == "application/json"
+        assert openapi_request.content_type == "application/json"
 
     def test_multiple_values(self, request_factory, request):
         request = request_factory(
@@ -60,7 +60,7 @@ def test_multiple_values(self, request_factory, request):
         assert openapi_request.host_url == "http://localhost"
         assert openapi_request.path == "/"
         assert openapi_request.body == prepared.body
-        assert openapi_request.mimetype == "application/json"
+        assert openapi_request.content_type == "application/json"
 
     def test_url_rule(self, request_factory, request):
         request = request_factory("GET", "/browse/12/", subdomain="kb")
@@ -87,7 +87,7 @@ def test_url_rule(self, request_factory, request):
         assert openapi_request.host_url == "http://localhost"
         assert openapi_request.path == "/browse/12/"
         assert openapi_request.body == prepared.body
-        assert openapi_request.mimetype == "application/json"
+        assert openapi_request.content_type == "application/json"
 
     def test_hash_param(self, request_factory, request):
         request = request_factory("GET", "/browse/#12", subdomain="kb")
@@ -114,7 +114,7 @@ def test_hash_param(self, request_factory, request):
         assert openapi_request.host_url == "http://localhost"
         assert openapi_request.path == "/browse/#12"
         assert openapi_request.body == prepared.body
-        assert openapi_request.mimetype == "application/json"
+        assert openapi_request.content_type == "application/json"
 
     def test_content_type_with_charset(self, request_factory, request):
         request = request_factory(
@@ -141,4 +141,6 @@ def test_content_type_with_charset(self, request_factory, request):
         assert openapi_request.host_url == "http://localhost"
         assert openapi_request.path == "/"
         assert openapi_request.body == prepared.body
-        assert openapi_request.mimetype == "application/json"
+        assert (
+            openapi_request.content_type == "application/json; charset=utf-8"
+        )