export a handle style decorator

[trim21]

I'm using handle (async def handle(request) -> Response) in my aiohttp application.

it looks like currently it only support class based view, not request.

I try a code snippet to make it work: (most of code are copied from inner)

from asyncio import iscoroutinefunction
from functools import update_wrapper
from typing import Callable, Iterable

from aiohttp import web
from aiohttp.web_response import json_response, StreamResponse
from aiohttp_pydantic.injectors import (
    CONTEXT, AbstractInjector, _parse_func_signature, MatchInfoGetter, BodyGetter, QueryGetter, HeadersGetter,
)
from pydantic import ValidationError

async def on_validation_error(exception: ValidationError, context: CONTEXT) -> StreamResponse:
    """
    This method is a hook to intercept ValidationError.

    This hook can be redefined to return a custom HTTP response error.
    The exception is a pydantic.ValidationError and the context is "body",
    "headers", "path" or "query string"
    """
    errors = exception.errors()
    for error in errors:
        error["in"] = context

    return json_response(data=errors, status=400)

def parse_func_signature(func: Callable) -> Iterable[AbstractInjector]:
    path_args, body_args, qs_args, header_args, defaults = _parse_func_signature(func)
    injectors = []

    def default_value(args: dict) -> dict:
        """
        Returns the default values of args.
        """
        return {name: defaults[name] for name in args if name in defaults}

    if path_args:
        injectors.append(MatchInfoGetter(path_args, default_value(path_args)))
    if body_args:
        injectors.append(BodyGetter(body_args, default_value(body_args)))
    if qs_args:
        injectors.append(QueryGetter(qs_args, default_value(qs_args)))
    if header_args:
        injectors.append(HeadersGetter(header_args, default_value(header_args)))
    return injectors

def decorator(handler):
    """
    Decorator to unpack the query string, route path, body and http header in
    the parameters of the web handler regarding annotations.
    """

    injectors = parse_func_signature(handler)

    async def wrapped_handler(request):
        args = []
        kwargs = {}
        for injector in injectors:
            try:
                if iscoroutinefunction(injector.inject):
                    await injector.inject(request, args, kwargs)
                else:
                    injector.inject(request, args, kwargs)
            except ValidationError as error:
                return await on_validation_error(error, injector.context)

        return await handler(*args, **kwargs)

    update_wrapper(wrapped_handler, handler)
    return wrapped_handler

@decorator
async def get(id: int = None, /, with_comments: bool = False, *, user_agent: str = None):
    return web.json_response(
        {
            'id': id,
            'UA': user_agent,
            'with_comments': with_comments,
        }
    )

app = web.Application()
app.router.add_get('/{id}', get)

if __name__ == '__main__':
    web.run_app(app, port=9092)

there is another sugestion:

all *Getter.inject can be asynchronous function, and just call await injector.inject(request, args, kwargs) without iscoroutinefunction call. ( and put BodyGetter at last )

[Maillol]

Good job! Have a decorator for HTTP handler function is a great feature but the handler decorator should works with aiohttp_pydantic.oas.view.generate_oas(). and the decorator should take the on_validation_error hook in optional parameter.

Do you want to try to create a Pull requests ?

[trim21]

Sorry to say that, but I'm lazy to create a PR for this...

And I don't use this OAS feature.

[Maillol]

No problem, I try to do that the next week. Thanks for your message it's always good to know how people use aiohttp-pydantic or what's it need.

[n1k0din]

No problem, I try to do that the next week.

@Maillol Is it now possible to use decorator instead of class based view?

[Maillol]

You can try this branch.

https://github.com/Maillol/aiohttp-pydantic/tree/add-decorator-to-use-with-web-handler-fonction

from aiohttp_pydantic.decorator import inject_params

@inject_params
async def get(request, name: str = "") -> r200[list[Friend]]:

# request parameter is optional

@inject_params
async def get(name: str = "") -> r200[list[Friend]]:

# You can add custom on_validation_error function

async def on_validation_error(exception: ValidationError, context: CONTEXT) -> StreamResponse:
    ...

@inject_params(on_validation_error=on_validation_error)
async def get(name: str = "") -> r200[list[Friend]]:
    ...