Moved route function parameters to type annotation and still support assignment pattern

Author: eadwinCode

Makefile

@@ -26,7 +26,7 @@ fmt format:clean ## Run code formatters
 	black ellar tests examples
 	ruff check --fix ellar tests examples
 
-test: ## Run tests
+test:clean ## Run tests
 	pytest
 
 test-cov: ## Run tests with coverage

ellar/common/__init__.py

@@ -60,20 +60,13 @@
 )
 from .params.decorators import (
     Body,
-    Context,
     Cookie,
     File,
     Form,
     Header,
-    Host,
-    Http,
+    Inject,
     Path,
-    Provide,
     Query,
-    Req,
-    Res,
-    Session,
-    Ws,
     WsBody,
 )
 from .params.params import ParamFieldInfo as Param
@@ -130,6 +123,7 @@
     "ModuleRouter",
     "render",
     "Module",
+    "Inject",
     "UseGuards",
     "Param",
     "ParamTypes",
@@ -154,19 +148,11 @@
     "Path",
     "Query",
     "WsBody",
-    "Context",
-    "Provide",
-    "Req",
-    "Ws",
     "middleware",
     "exception_handler",
     "serializer_filter",
     "template_filter",
     "template_global",
-    "Res",
-    "Session",
-    "Host",
-    "Http",
     "UploadFile",
     "file",
     "extra_args",

ellar/common/params/__init__.py

@@ -4,13 +4,15 @@
     RequestEndpointArgsModel,
     WebsocketEndpointArgsModel,
 )
+from .decorators import add_default_resolver
 from .resolvers import (
     BaseConnectionParameterResolver,
     IRouteParameterResolver,
     NonParameterResolver,
 )
 
 __all__ = [
+    "add_default_resolver",
     "WebsocketEndpointArgsModel",
     "RequestEndpointArgsModel",
     "ExtraEndpointArg",

ellar/common/params/args/__init__.py

@@ -1,4 +1,4 @@
-from .base import EndpointArgsModel, add_default_resolver
+from .base import EndpointArgsModel
 from .extra_args import ExtraEndpointArg
 from .request_model import RequestEndpointArgsModel
 from .websocket_model import WebsocketEndpointArgsModel
@@ -8,5 +8,4 @@
     "RequestEndpointArgsModel",
     "WebsocketEndpointArgsModel",
     "EndpointArgsModel",
-    "add_default_resolver",
 ]

ellar/common/params/args/base.py

@@ -15,16 +15,8 @@
 from pydantic.fields import FieldInfo, ModelField
 from pydantic.typing import ForwardRef, evaluate_forwardref  # type:ignore
 from pydantic.utils import Representation, lenient_issubclass
-from starlette.background import BackgroundTasks
 from starlette.convertors import Convertor
-from starlette.requests import (
-    HTTPConnection as StarletteHTTPConnection,
-)
-from starlette.requests import (
-    Request as StarletteRequest,
-)
-from starlette.responses import Response
-from starlette.websockets import WebSocket as StarletteWebSocket
+from typing_extensions import Annotated, get_args, get_origin
 
 from .. import params
 from ..helpers import is_scalar_field, is_scalar_sequence_field
@@ -33,14 +25,6 @@
     IRouteParameterResolver,
     NonParameterResolver,
 )
-from ..resolvers.non_parameter import (
-    BackgroundTasksParameter,
-    ConnectionParam,
-    ExecutionContextParameter,
-    RequestParameter,
-    ResponseRequestParam,
-    WebSocketParameter,
-)
 from .extra_args import ExtraEndpointArg
 from .factory import get_parameter_field
 from .resolver_generators import (
@@ -50,21 +34,6 @@
     QueryHeaderResolverGenerator,
 )
 
-DEFAULT_RESOLVERS: t.Dict[t.Type, t.Type[NonParameterResolver]] = {
-    StarletteRequest: RequestParameter,
-    StarletteWebSocket: WebSocketParameter,
-    Response: ResponseRequestParam,
-    StarletteHTTPConnection: ConnectionParam,
-    IExecutionContext: ExecutionContextParameter,
-    BackgroundTasks: BackgroundTasksParameter,
-}
-
-
-def add_default_resolver(
-    type_identifier: t.Type, resolver_type: t.Type[NonParameterResolver]
-) -> None:  # pragma: no cover
-    DEFAULT_RESOLVERS.update({type_identifier: resolver_type})
-
 
 class EndpointArgsModel:
     _bulk_resolvers_generators = {
@@ -185,27 +154,30 @@ def compute_route_parameter_list(
         self, body_field_class: t.Type[FieldInfo] = params.BodyFieldInfo
     ) -> None:
         for param_name, param in self.endpoint_signature.parameters.items():
+            param_name, param_default, param_annotation, param_kind = (
+                param.name,
+                param.default,
+                param.annotation,
+                param.kind,
+            )
+            param_annotation, param_default = self._get_annotation_type_and_default(
+                param_annotation, param_default
+            )
+
             if (
-                param.kind == param.VAR_KEYWORD
-                or param.kind == param.VAR_POSITIONAL
+                param_kind == param.VAR_KEYWORD
+                or param_kind == param.VAR_POSITIONAL
                 or (
-                    param.name == "self" and param.annotation == inspect.Parameter.empty
+                    param_name == "self" and param_annotation == inspect.Parameter.empty
                 )
             ):
                 # Skipping **kwargs, *args, self
                 continue
 
-            if self._add_non_pydantic_field_to_dependency(
-                param_name=param.name,
-                param_default=param.default,
-                param_annotation=param.annotation,
-            ):
-                continue
-
             if self._add_non_field_param_to_dependency(
-                param_name=param.name,
-                param_default=param.default,
-                param_annotation=param.annotation,
+                param_name=param_name,
+                param_default=param_default,
+                param_annotation=param_annotation,
             ):
                 continue
 
@@ -215,8 +187,8 @@ def compute_route_parameter_list(
                 else:
                     ignore_default = True
                 param_field = get_parameter_field(
-                    param_default=param.default,
-                    param_annotation=param.annotation,
+                    param_default=param_default,
+                    param_annotation=param_annotation,
                     param_name=param_name,
                     default_field_info=params.PathFieldInfo,
                     ignore_default=ignore_default,
@@ -233,8 +205,8 @@ def compute_route_parameter_list(
                     else params.QueryFieldInfo,
                 )
                 param_field = get_parameter_field(
-                    param_default=param.default,
-                    param_annotation=param.annotation,
+                    param_default=param_default,
+                    param_annotation=param_annotation,
                     default_field_info=default_field_info,
                     param_name=param_name,
                     body_field_class=body_field_class,
@@ -332,14 +304,38 @@ async def resolve_dependencies(
     def compute_extra_route_args(self) -> None:
         self._add_extra_route_args(*self._extra_endpoint_args)
 
+    def _get_annotation_type_and_default(
+        self, param_annotation: t.Any, default_param_default: t.Any
+    ) -> t.Tuple:
+        if get_origin(param_annotation) is Annotated:
+            annotated_args = get_args(param_annotation)
+            if len(annotated_args) == 2:
+                return annotated_args
+            else:
+                raise ImproperConfiguration(
+                    f"Cannot specify multiple `Annotated` Ellar arguments for {annotated_args!r}"
+                )
+
+        return param_annotation, default_param_default
+
     def _add_extra_route_args(
         self, *extra_operation_args: ExtraEndpointArg, key: t.Optional[str] = None
     ) -> None:
         for param in extra_operation_args:
+            param_name, param_default, param_annotation = (
+                param.name,
+                param.default,
+                param.annotation,
+            )
+
+            param_annotation, param_default = self._get_annotation_type_and_default(
+                param_annotation, param_default
+            )
+
             if self._add_non_field_param_to_dependency(
-                param_name=param.name,
-                param_default=param.default,
-                param_annotation=param.annotation,
+                param_name=param_name,
+                param_default=param_default,
+                param_annotation=param_annotation,
                 key=key,
             ):
                 continue
@@ -379,13 +375,3 @@ def __copy__(
 
     def build_body_field(self) -> None:  # pragma: no cover
         raise NotImplementedError
-
-    def _add_non_pydantic_field_to_dependency(
-        self, param_name: str, param_default: t.Any, param_annotation: t.Any
-    ) -> bool:
-        resolver_class = DEFAULT_RESOLVERS.get(param_annotation)
-        if resolver_class and param_default == inspect.Parameter.empty:
-            _inject = resolver_class()(param_name, param_annotation)
-            self._computation_models[_inject.in_].append(_inject)
-            return True
-        return False

ellar/common/params/decorators/__init__.py

@@ -0,0 +1,98 @@
+import typing as t
+
+from typing_extensions import Annotated
+
+from . import models as param_functions
+from .inject import InjectShortcut, add_default_resolver
+
+__all__ = [
+    "add_default_resolver",
+    "Body",
+    "Cookie",
+    "File",
+    "Form",
+    "Header",
+    "Path",
+    "Query",
+    "WsBody",
+    "Inject",
+]
+
+
+class _ParamShortcut:
+    def __init__(self, base_func: t.Callable) -> None:
+        self._base_func = base_func
+
+    def __call__(self, *args: t.Any, **kwargs: t.Any) -> t.Any:
+        return self._base_func(*args, **kwargs)
+
+    def __getitem__(self, args: t.Any) -> t.Any:
+        if isinstance(args, tuple):
+            return Annotated[args[0], self._base_func(**args[1])]
+        return Annotated[args, self._base_func()]
+
+    @classmethod
+    def P(
+        cls,
+        default: t.Any = ...,
+        *,
+        alias: t.Optional[str] = None,
+        title: t.Optional[str] = None,
+        description: t.Optional[str] = None,
+        gt: t.Optional[float] = None,
+        ge: t.Optional[float] = None,
+        lt: t.Optional[float] = None,
+        le: t.Optional[float] = None,
+        min_length: t.Optional[int] = None,
+        max_length: t.Optional[int] = None,
+        regex: t.Optional[str] = None,
+        example: t.Any = None,
+        examples: t.Optional[t.Dict[str, t.Any]] = None,
+        deprecated: t.Optional[bool] = None,
+        include_in_schema: bool = True,
+        **extra: t.Any,
+    ) -> t.Dict[str, t.Any]:
+        """Arguments for Body, Query, Header, Cookie, etc."""
+        return dict(
+            default=default,
+            alias=alias,
+            title=title,
+            description=description,
+            gt=gt,
+            ge=ge,
+            lt=lt,
+            le=le,
+            min_length=min_length,
+            max_length=max_length,
+            regex=regex,
+            example=example,
+            examples=examples,
+            deprecated=deprecated,
+            include_in_schema=include_in_schema,
+            **extra,
+        )
+
+
+if t.TYPE_CHECKING:  # pragma: nocover
+    # mypy cheats
+    T = t.TypeVar("T")
+    Body = Annotated[T, param_functions.Body()]
+    Cookie = Annotated[T, param_functions.Cookie()]
+    File = Annotated[T, param_functions.File()]
+    Form = Annotated[T, param_functions.Form()]
+    Header = Annotated[T, param_functions.Header()]
+    Path = Annotated[T, param_functions.Path()]
+    Query = Annotated[T, param_functions.Query()]
+    WsBody = Annotated[T, param_functions.WsBody()]
+    Inject = Annotated[T, t.Any]
+
+else:
+    Body = _ParamShortcut(param_functions.Body)
+    Cookie = _ParamShortcut(param_functions.Cookie)
+    File = _ParamShortcut(param_functions.File)
+    Form = _ParamShortcut(param_functions.Form)
+    Header = _ParamShortcut(param_functions.Header)
+    Path = _ParamShortcut(param_functions.Path)
+    Query = _ParamShortcut(param_functions.Query)
+    WsBody = _ParamShortcut(param_functions.WsBody)
+    Inject = InjectShortcut