🔨 Class-Based Parsers and Serializers (#96)

.flake8

@@ -7,5 +7,6 @@ docstring-convention = google
 per-file-ignores =
     tests/*:S101,D100,D205,D415,S106
     __init__.py:F401
+    typing.py:F401
 rst-roles = class,const,func,meth,mod,ref
 rst-directives = deprecated

docs/installation.rst

@@ -21,7 +21,7 @@ Install Flask-Jeroboam
          $ pip install flask-jeroboam
 
 
-**Flask-Jeroboam** is now installed along with its two direct dependencies, `Flask`_and `Pydantic`_ as well as their own dependencies tree. :ref:`see <Dependencies>`
+**Flask-Jeroboam** is now installed along with its two direct dependencies, `Flask`_ and `Pydantic`__ as well as their own dependencies tree. :ref:`see <Dependencies>`
 
 If you already have dependency management in Python figured out, skip our next section :ref:`About Dependency Management`. If not, :ref:`check it out <About Dependency Management>` before moving on to the :doc:`Getting Started Guide </getting_started>` or go to the
 :doc:`Documentation Overview </index>`.

flask_jeroboam/__init__.py

@@ -1,2 +1,2 @@
-from .blueprint import APIBlueprint
+from .jeroboam import APIBlueprint
 from .jeroboam import Jeroboam

flask_jeroboam/_parser.py

@@ -3,12 +3,20 @@
 import typing as t
 from enum import Enum
 from functools import wraps
+from typing import Callable
+from typing import List
+from typing import Type
 
 from flask import request
+from flask.globals import current_app
 from pydantic import BaseModel
 from typing_extensions import ParamSpec
 
 from flask_jeroboam.exceptions import InvalidRequest
+from flask_jeroboam.typing import JeroboamResponseReturnValue
+from flask_jeroboam.typing import JeroboamRouteCallable
+
+from .utils import get_typed_signature
 
 
 F = t.TypeVar("F", bound=t.Callable[..., t.Any])
@@ -30,66 +38,90 @@ class MethodEnum(str, Enum):
 pattern = r"(.*)\[(.+)\]$"
 
 
-def _parse_input(model: t.Type[BaseModel], **kwargs: ParamSpec) -> BaseModel:
-    try:
-        return model(**kwargs)
-    except ValueError as e:
-        raise InvalidRequest(msg=str(e)) from e
-
-
-def _simple_parse_input(type_: T, payload: dict, key: str) -> T:
-    try:
-        return type_(payload.get(key, None))
-    except ValueError as e:
-        raise InvalidRequest(msg=str(e)) from e
-
-
-def _rename_keys(location: dict, pattern: str) -> dict:
-    renamings = []
-    for key, value in location.items():
-        match = re.match(pattern, key)
-        if len(value) == 1 and match is None:
-            location[key] = value[0]
-        elif match is not None:
-            new_key = f"{match[1]}[]"
-            new_value = {match[2]: value[0]}
-            renamings.append((key, new_key, new_value))
-    for key, new_key, new_value in renamings:
-        if new_key not in location:
-            location[new_key] = [new_value]
-        else:
-            location[new_key].append(new_value)
-        del location[key]
-    return location
-
-
-def parser(
-    method: str, rule: str
-) -> t.Callable[[t.Callable[..., R]], t.Callable[..., R]]:
-    """Parametrize Decorator for parsing request data."""
-
-    def parser_decorator(func: t.Callable[..., R]) -> t.Callable[..., R]:
+class Parser:
+    """A Parser Class for Flask-Jeroboam."""
+
+    def __init__(self, func: Callable, methods: List[str], rule: str):
+        self.typed_params = get_typed_signature(func)
+        self.methods = methods
+        self.rule = rule
+
+    def __bool__(self) -> bool:
+        return bool(self.typed_params.parameters)
+
+    def __call__(self, func: JeroboamRouteCallable) -> JeroboamRouteCallable:
+        """It injects inbound parsed and validated data into the view function."""
+
         @wraps(func)
-        def wrapper(*args, **kwargs):
-            if method == MethodEnum.GET:
-                location = dict(request.args.lists())
-                location = _rename_keys(location, pattern)
-            elif method == MethodEnum.POST:
-                location = dict(request.form)
-                if request.data:
-                    location.update(dict(json.loads(request.data)))
-                location.update(dict(request.files))  # type: ignore
-            else:
-                location = {}
-            for arg_name, arg_type in func.__annotations__.items():
-                if getattr(arg_type, "__origin__", None) == t.Union:
-                    arg_type = arg_type.__args__[0]
-                if issubclass(arg_type, BaseModel):
-                    kwargs[arg_name] = _parse_input(arg_type, **location)
-                elif arg_name not in rule:
-                    kwargs[arg_name] = _simple_parse_input(arg_type, location, arg_name)
-            return func(*args, **kwargs)
+        def wrapper(*args, **kwargs) -> JeroboamResponseReturnValue:
+            location = self._parse_incoming_request_data()
+            kwargs = self._validate_inbound_data(location, kwargs)
+            return current_app.ensure_sync(func)(*args, **kwargs)
 
         return wrapper
 
-    return parser_decorator
+    def _parse_incoming_request_data(self) -> dict:
+        """Getting the Data out of the Request Object."""
+        if MethodEnum.GET in self.methods:
+            location = dict(request.args.lists())
+            location = self._rename_query_params_keys(location, pattern)
+        elif MethodEnum.POST in self.methods:
+            location = dict(request.form.lists())
+            location = self._rename_query_params_keys(location, pattern)
+            if request.data:
+                # TODO: on.3.8.drop location |= dict(json.loads(request.data))
+                location.update(dict(json.loads(request.data)))
+            # TODO: on.3.8.drop location |= dict(request.files)  # type: ignore
+            location.update(dict(request.files))  # type: ignore
+        else:  # pragma: no cover
+            # TODO: Statement cannot be reached at this point.
+            location = {}
+        return location
+
+    def _validate_inbound_data(self, location, kwargs) -> dict:
+        """Getting the Data out of the Request Object."""
+        for arg_name, typed_param in self.typed_params.parameters.items():
+            if getattr(typed_param.annotation, "__origin__", None) == t.Union:
+                kwargs[arg_name] = self._validate_input(
+                    typed_param.annotation.__args__[0], **location
+                )
+            elif issubclass(typed_param.annotation, BaseModel):
+                kwargs[arg_name] = self._validate_input(
+                    typed_param.annotation, **location
+                )
+            elif arg_name not in self.rule:
+                kwargs[arg_name] = self._simple_validate_input(
+                    typed_param.annotation, location, arg_name
+                )
+        return kwargs
+
+    def _validate_input(self, model: Type[BaseModel], **kwargs: ParamSpec) -> BaseModel:
+        try:
+            return model(**kwargs)
+        except ValueError as e:
+            raise InvalidRequest(msg=str(e)) from e
+
+    def _simple_validate_input(self, type_: T, payload: dict, key: str) -> T:
+        try:
+            return type_(payload.get(key, None))
+        except ValueError as e:
+            raise InvalidRequest(msg=str(e)) from e
+
+    def _rename_query_params_keys(self, inbound_dict: dict, pattern: str) -> dict:
+        """Rename keys in a dictionary."""
+        renamings = []
+        for key, value in inbound_dict.items():
+            match = re.match(pattern, key)
+            if len(value) == 1 and match is None:
+                inbound_dict[key] = value[0]
+            elif match is not None:
+                new_key = f"{match[1]}[]"
+                new_value = {match[2]: value[0]}
+                renamings.append((key, new_key, new_value))
+        for key, new_key, new_value in renamings:
+            if new_key not in inbound_dict:
+                inbound_dict[new_key] = [new_value]
+            else:
+                inbound_dict[new_key].append(new_value)
+            del inbound_dict[key]
+        return inbound_dict

flask_jeroboam/_serializer.py

@@ -1,50 +1,84 @@
 import traceback
-import typing as t
 from functools import wraps
+from typing import Any
 from typing import Callable
+from typing import Dict
+from typing import Optional
 from typing import TypeVar
 
 from flask import Response
+from flask.globals import current_app
 from pydantic import BaseModel
 from typing_extensions import ParamSpec
 
 from .exceptions import ServerError
+from .typing import JeroboamResponseReturnValue
+from .typing import JeroboamRouteCallable
+from .typing import ResponseModel
+
+
+# from .typing import TypedParams
+# from .utils import get_typed_return_annotation
 
 
 P = ParamSpec("P")
 R = TypeVar("R")
 
 
-def _prepare_response(result: BaseModel, status: int = 200) -> Response:
-    """Wraps simple Response initialisation."""
-    return Response(result.json(), mimetype="application/json", status=status)
+class Serializer:
+    """A Serializer Class for Flask-Jeroboam."""
+
+    def __init__(self, func: Callable, options: Dict[str, Any]):
+        self.response_model = self.get_response_model(func, options)
+
+    def get_response_model(
+        self, func: Callable, options: Dict[str, Any]
+    ) -> Optional[ResponseModel]:
+        """Extract the Response Model from view function.
 
+        It takes it either explicitly from the options
+        or from return type of the view function if it is
+        a pydantic BaseModel.
+        """
+        response_model = options.pop("response_model", None)
+        """if response_model is None:
+            return_annotation = get_typed_return_annotation(func)
+            if issubclass(return_annotation, BaseModel):
+                response_model = return_annotation """
+        return response_model
 
-def serializer(response_model: t.Type[BaseModel], status_code: int = 200):
-    """Parameterized decorator for view functions."""
+    def __bool__(self) -> bool:
+        """Return True if the Serializer has a Response Model."""
+        return bool(self.response_model)
+
+    def _prepare_response(self, result: BaseModel, status: int = 200) -> Response:
+        """Wraps simple Response initialisation."""
+        return Response(result.json(), status=status, mimetype="application/json")
+
+    def __call__(
+        self, func: JeroboamRouteCallable, status_code: int
+    ) -> JeroboamRouteCallable:
+        """Return a view funcion."""
 
-    def serialize_decorator(
-        func: Callable[P, t.Union[R, Response]]
-    ) -> Callable[P, t.Union[R, Response]]:
         @wraps(func)
-        def inner(*args: P.args, **kwargs: P.kwargs) -> t.Union[R, Response]:
-            response = func(*args, **kwargs)
-            if isinstance(response, dict):
+        def wrapper(*args, **kwargs) -> JeroboamResponseReturnValue:
+            response = current_app.ensure_sync(func)(*args, **kwargs)
+            if self.response_model is None:  # pragma: no cover
+                return response
+            elif isinstance(response, dict):
                 try:
-                    validated_response = response_model(**response)
+                    validated_response = self.response_model(**response)
                 except ValueError as e:
                     raise ServerError(
                         msg="Internal server error",
                         error=e,
                         trace=traceback.format_exc(),
                         context=f"Trying to validate result with value {response}.",
                     ) from e
-                return _prepare_response(validated_response, status_code)
-            elif isinstance(response, response_model):
-                return _prepare_response(response, status_code)
+                return self._prepare_response(validated_response, status_code)
+            elif isinstance(response, self.response_model):
+                return self._prepare_response(response, status_code)
             else:
                 return response
 
-        return inner
-
-    return serialize_decorator
+        return wrapper

flask_jeroboam/jeroboam.py

@@ -1,7 +1,37 @@
-"""The Flask Object with augmented functionality around route registration."""
+"""The Flask Object with augmented functionality around route registration.
+
+Here we overide the route method of the Flask object to use our custom implementation.
+This allow us to introduce new functionality to the route registration process.
+"""
+from typing import Any
+from typing import Callable
+
 from flask import Flask
+from flask.blueprints import Blueprint
+from flask.scaffold import Scaffold
+from typing_extensions import TypeVar
+
+from .typing import JeroboamRouteCallable
+from .view import JeroboamViewFunction
+
+
+R = TypeVar("R", bound=Any)
+
 
-from flask_jeroboam._route import route
+def route_override(
+    self: Scaffold, rule: str, **options: Any
+) -> Callable[[JeroboamRouteCallable], JeroboamRouteCallable]:
+    """Route Registration Override."""
+
+    def decorator(func: JeroboamRouteCallable) -> JeroboamRouteCallable:
+        route = JeroboamViewFunction(rule, func, options)
+
+        self.add_url_rule(
+            rule, route.endpoint, route.as_view, **options  # type: ignore
+        )
+        return func
+
+    return decorator
 
 
 class Jeroboam(Flask):
@@ -11,4 +41,10 @@ class Jeroboam(Flask):
     route decorator.
     """
 
-    route = route  # type: ignore
+    route = route_override  # type: ignore[assignment]
+
+
+class APIBlueprint(Blueprint):
+    """Regular Blueprint with extra behavior on route definition."""
+
+    route = route_override  # type: ignore[assignment]