Static types with mypy

Author: p1c2u

openapi_core/casting/schemas/casters.py

@@ -1,49 +1,59 @@
+from typing import TYPE_CHECKING
+from typing import Any
+from typing import Callable
+from typing import List
+
+from openapi_core.casting.schemas.datatypes import CasterCallable
 from openapi_core.casting.schemas.exceptions import CastError
+from openapi_core.spec import Spec
+
+if TYPE_CHECKING:
+    from openapi_core.casting.schemas.factories import SchemaCastersFactory
 
 
 class BaseSchemaCaster:
-    def __init__(self, schema):
+    def __init__(self, schema: Spec):
         self.schema = schema
 
-    def __call__(self, value):
+    def __call__(self, value: Any) -> Any:
         if value is None:
             return value
 
         return self.cast(value)
 
-    def cast(self, value):
+    def cast(self, value: Any) -> Any:
         raise NotImplementedError
 
 
 class CallableSchemaCaster(BaseSchemaCaster):
-    def __init__(self, schema, caster_callable):
+    def __init__(self, schema: Spec, caster_callable: CasterCallable):
         super().__init__(schema)
         self.caster_callable = caster_callable
 
-    def cast(self, value):
+    def cast(self, value: Any) -> Any:
         try:
             return self.caster_callable(value)
         except (ValueError, TypeError):
             raise CastError(value, self.schema["type"])
 
 
 class DummyCaster(BaseSchemaCaster):
-    def cast(self, value):
+    def cast(self, value: Any) -> Any:
         return value
 
 
 class ComplexCaster(BaseSchemaCaster):
-    def __init__(self, schema, casters_factory):
+    def __init__(self, schema: Spec, casters_factory: "SchemaCastersFactory"):
         super().__init__(schema)
         self.casters_factory = casters_factory
 
 
 class ArrayCaster(ComplexCaster):
     @property
-    def items_caster(self):
+    def items_caster(self) -> BaseSchemaCaster:
         return self.casters_factory.create(self.schema / "items")
 
-    def cast(self, value):
+    def cast(self, value: Any) -> List[Any]:
         try:
             return list(map(self.items_caster, value))
         except (ValueError, TypeError):

openapi_core/casting/schemas/datatypes.py

@@ -0,0 +1,4 @@
+from typing import Any
+from typing import Callable
+
+CasterCallable = Callable[[Any], Any]

openapi_core/casting/schemas/exceptions.py

@@ -10,5 +10,5 @@ class CastError(OpenAPIError):
     value: str
     type: str
 
-    def __str__(self):
+    def __str__(self) -> str:
         return f"Failed to cast value to {self.type} type: {self.value}"

openapi_core/casting/schemas/factories.py

@@ -1,6 +1,11 @@
+from typing import Dict
+
 from openapi_core.casting.schemas.casters import ArrayCaster
+from openapi_core.casting.schemas.casters import BaseSchemaCaster
 from openapi_core.casting.schemas.casters import CallableSchemaCaster
 from openapi_core.casting.schemas.casters import DummyCaster
+from openapi_core.casting.schemas.datatypes import CasterCallable
+from openapi_core.spec import Spec
 from openapi_core.util import forcebool
 
 
@@ -11,7 +16,7 @@ class SchemaCastersFactory:
         "object",
         "any",
     ]
-    PRIMITIVE_CASTERS = {
+    PRIMITIVE_CASTERS: Dict[str, CasterCallable] = {
         "integer": int,
         "number": float,
         "boolean": forcebool,
@@ -20,7 +25,7 @@ class SchemaCastersFactory:
         "array": ArrayCaster,
     }
 
-    def create(self, schema):
+    def create(self, schema: Spec) -> BaseSchemaCaster:
         schema_type = schema.getkey("type", "any")
 
         if schema_type in self.DUMMY_CASTERS:

openapi_core/deserializing/media_types/datatypes.py

@@ -0,0 +1,4 @@
+from typing import Any
+from typing import Callable
+
+DeserializerCallable = Callable[[Any], Any]

openapi_core/deserializing/media_types/deserializers.py

@@ -1,30 +1,37 @@
 import warnings
+from typing import Any
+from typing import Callable
 
+from openapi_core.deserializing.media_types.datatypes import (
+    DeserializerCallable,
+)
 from openapi_core.deserializing.media_types.exceptions import (
     MediaTypeDeserializeError,
 )
 
 
 class BaseMediaTypeDeserializer:
-    def __init__(self, mimetype):
+    def __init__(self, mimetype: str):
         self.mimetype = mimetype
 
-    def __call__(self, value):
+    def __call__(self, value: Any) -> Any:
         raise NotImplementedError
 
 
 class UnsupportedMimetypeDeserializer(BaseMediaTypeDeserializer):
-    def __call__(self, value):
+    def __call__(self, value: Any) -> Any:
         warnings.warn(f"Unsupported {self.mimetype} mimetype")
         return value
 
 
 class CallableMediaTypeDeserializer(BaseMediaTypeDeserializer):
-    def __init__(self, mimetype, deserializer_callable):
+    def __init__(
+        self, mimetype: str, deserializer_callable: DeserializerCallable
+    ):
         self.mimetype = mimetype
         self.deserializer_callable = deserializer_callable
 
-    def __call__(self, value):
+    def __call__(self, value: Any) -> Any:
         try:
             return self.deserializer_callable(value)
         except (ValueError, TypeError, AttributeError):

openapi_core/deserializing/media_types/exceptions.py

@@ -10,7 +10,7 @@ class MediaTypeDeserializeError(DeserializeError):
     mimetype: str
     value: str
 
-    def __str__(self):
+    def __str__(self) -> str:
         return (
             "Failed to deserialize value with {mimetype} mimetype: {value}"
         ).format(value=self.value, mimetype=self.mimetype)

openapi_core/deserializing/media_types/factories.py

@@ -1,5 +1,15 @@
 from json import loads
+from typing import Any
+from typing import Callable
+from typing import Dict
+from typing import Optional
 
+from openapi_core.deserializing.media_types.datatypes import (
+    DeserializerCallable,
+)
+from openapi_core.deserializing.media_types.deserializers import (
+    BaseMediaTypeDeserializer,
+)
 from openapi_core.deserializing.media_types.deserializers import (
     CallableMediaTypeDeserializer,
 )
@@ -12,26 +22,31 @@
 
 class MediaTypeDeserializersFactory:
 
-    MEDIA_TYPE_DESERIALIZERS = {
+    MEDIA_TYPE_DESERIALIZERS: Dict[str, DeserializerCallable] = {
         "application/json": loads,
         "application/x-www-form-urlencoded": urlencoded_form_loads,
         "multipart/form-data": data_form_loads,
     }
 
-    def __init__(self, custom_deserializers=None):
+    def __init__(
+        self,
+        custom_deserializers: Optional[Dict[str, DeserializerCallable]] = None,
+    ):
         if custom_deserializers is None:
             custom_deserializers = {}
         self.custom_deserializers = custom_deserializers
 
-    def create(self, mimetype):
+    def create(self, mimetype: str) -> BaseMediaTypeDeserializer:
         deserialize_callable = self.get_deserializer_callable(mimetype)
 
         if deserialize_callable is None:
             return UnsupportedMimetypeDeserializer(mimetype)
 
         return CallableMediaTypeDeserializer(mimetype, deserialize_callable)
 
-    def get_deserializer_callable(self, mimetype):
+    def get_deserializer_callable(
+        self, mimetype: str
+    ) -> Optional[DeserializerCallable]:
         if mimetype in self.custom_deserializers:
             return self.custom_deserializers[mimetype]
         return self.MEDIA_TYPE_DESERIALIZERS.get(mimetype)

openapi_core/deserializing/media_types/util.py

@@ -1,13 +1,16 @@
 from email.parser import Parser
+from typing import Any
+from typing import Dict
+from typing import Union
 from urllib.parse import parse_qsl
 
 
-def urlencoded_form_loads(value):
+def urlencoded_form_loads(value: Any) -> Dict[str, Any]:
     return dict(parse_qsl(value))
 
 
-def data_form_loads(value):
-    if issubclass(type(value), bytes):
+def data_form_loads(value: Union[str, bytes]) -> Dict[str, Any]:
+    if isinstance(value, bytes):
         value = value.decode("ASCII", errors="surrogateescape")
     parser = Parser()
     parts = parser.parsestr(value, headersonly=False)

openapi_core/deserializing/parameters/datatypes.py

@@ -0,0 +1,4 @@
+from typing import Callable
+from typing import List
+
+DeserializerCallable = Callable[[str], List[str]]

openapi_core/deserializing/parameters/deserializers.py

@@ -1,37 +1,49 @@
 import warnings
+from typing import Any
+from typing import Callable
+from typing import List
 
 from openapi_core.deserializing.exceptions import DeserializeError
+from openapi_core.deserializing.parameters.datatypes import (
+    DeserializerCallable,
+)
 from openapi_core.deserializing.parameters.exceptions import (
     EmptyQueryParameterValue,
 )
 from openapi_core.schema.parameters import get_aslist
 from openapi_core.schema.parameters import get_explode
+from openapi_core.spec import Spec
 
 
 class BaseParameterDeserializer:
-    def __init__(self, param_or_header, style):
+    def __init__(self, param_or_header: Spec, style: str):
         self.param_or_header = param_or_header
         self.style = style
 
-    def __call__(self, value):
+    def __call__(self, value: Any) -> Any:
         raise NotImplementedError
 
 
 class UnsupportedStyleDeserializer(BaseParameterDeserializer):
-    def __call__(self, value):
+    def __call__(self, value: Any) -> Any:
         warnings.warn(f"Unsupported {self.style} style")
         return value
 
 
 class CallableParameterDeserializer(BaseParameterDeserializer):
-    def __init__(self, param_or_header, style, deserializer_callable):
+    def __init__(
+        self,
+        param_or_header: Spec,
+        style: str,
+        deserializer_callable: DeserializerCallable,
+    ):
         super().__init__(param_or_header, style)
         self.deserializer_callable = deserializer_callable
 
         self.aslist = get_aslist(self.param_or_header)
         self.explode = get_explode(self.param_or_header)
 
-    def __call__(self, value):
+    def __call__(self, value: Any) -> Any:
         # if "in" not defined then it's a Header
         if "allowEmptyValue" in self.param_or_header:
             warnings.warn(

openapi_core/deserializing/parameters/exceptions.py

@@ -17,7 +17,7 @@ class ParameterDeserializeError(BaseParameterDeserializeError):
     style: str
     value: str
 
-    def __str__(self):
+    def __str__(self) -> str:
         return (
             "Failed to deserialize value of "
             f"{self.location} parameter with style {self.style}: {self.value}"
@@ -28,11 +28,11 @@ def __str__(self):
 class EmptyQueryParameterValue(BaseParameterDeserializeError):
     name: str
 
-    def __init__(self, name):
+    def __init__(self, name: str):
         super().__init__(location="query")
         self.name = name
 
-    def __str__(self):
+    def __str__(self) -> str:
         return (
             f"Value of {self.name} {self.location} parameter cannot be empty"
         )

openapi_core/deserializing/parameters/factories.py

@@ -1,5 +1,12 @@
 from functools import partial
+from typing import Dict
 
+from openapi_core.deserializing.parameters.datatypes import (
+    DeserializerCallable,
+)
+from openapi_core.deserializing.parameters.deserializers import (
+    BaseParameterDeserializer,
+)
 from openapi_core.deserializing.parameters.deserializers import (
     CallableParameterDeserializer,
 )
@@ -8,18 +15,19 @@
 )
 from openapi_core.deserializing.parameters.util import split
 from openapi_core.schema.parameters import get_style
+from openapi_core.spec import Spec
 
 
 class ParameterDeserializersFactory:
 
-    PARAMETER_STYLE_DESERIALIZERS = {
+    PARAMETER_STYLE_DESERIALIZERS: Dict[str, DeserializerCallable] = {
         "form": partial(split, separator=","),
         "simple": partial(split, separator=","),
         "spaceDelimited": partial(split, separator=" "),
         "pipeDelimited": partial(split, separator="|"),
     }
 
-    def create(self, param_or_header):
+    def create(self, param_or_header: Spec) -> BaseParameterDeserializer:
         style = get_style(param_or_header)
 
         if style not in self.PARAMETER_STYLE_DESERIALIZERS:

openapi_core/deserializing/parameters/util.py

@@ -1,2 +1,5 @@
-def split(value, separator=","):
+from typing import List
+
+
+def split(value: str, separator: str = ",") -> List[str]:
     return value.split(separator)