add support for general purpose middleware

[dominiquegarmier]

so a question was asked in discord about how to best reuse logic between fields (or queries and mutations). If the logic doesn't depend on arguments passed to the field we would use permission classes for this.

However if the logic depends on the inputs to that field there is currently, to my knowledge, no better way, than to outsource that logic into a function and call it at the beginning of each field. So i though about this and came up with this idea:

suppose we allowed an argument such as perhaps middleware to strawberry.field, which would be a list of functions, that get executed (in order), on the input. Forwarding the input (to the next middleware function or the field itself) if everything is fine, perhaps raising exceptions or even returning a response early. Importantly the input arguments should not be mutated as to allow for a predictable execution.

Perhaps a user would be able to define middleware by inheriting from some middleware class, where the user would have to implement a specific class method.

from strawberry.middleware import BaseMiddleware, MiddlewareResponse

@strawberry.type
class Result:
    message: str

@strawberry.input
class Input:
    password: str
    ...

class MyMiddleware(BaseMiddleware):
    def run(self, info: Info, input: Input) -> MiddlewareResponse[Result]:
        user = info.context['user']
        if is_lucky_user(user):
            return MiddlewareResponse(Result(message="foo"))  # this will get returned no other middleware will run
        elif user.password != input.password:
            raise ValueError("wrong password")  # an exception will get returned
        return  # nothing happens, next middleware or field will run normally

@strawberry.type
class Query
    @strawberry.field(middleware=[MyMiddleware])
    async def my_field(self, input: Input) -> Result: ...

run would have to be able to handle all arguments provided to my_field, which i think should be fine. But at the moment I mainly see the problem that the middleware has to return a specific Result type.

What are your thoughts on this? would this be a useful feature? Is there already a better way to do this?

[LeKSuS-04]

I was the one asking the original question on discord.

As I see it, run method of the middleware should have return type that fits query/mutation. So, with your example, it would look like this:

from strawberry.middleware import BaseMiddleware, MiddlewareResponse

@strawberry.type
class Result:
    message: str

@strawberry.type
class PossibleMiddlewareResult:
    something: str

@strawberry.type
class WrongPasswordError:
    details: str = 'Wrong password was provided'

@strawberry.input
class Input:
    password: str
    ...

MyQueryResult = strawberry.union(
    'MyQueryResult', (Result, PossibleMiddlewareResult, WrongPasswordError)
)

class MyMiddleware(BaseMiddleware):
    def run(self, info: Info, input: Input) -> MiddlewareResponse[PossibleMiddlewareResult, WrongPasswordError]:
        user = info.context['user']
        if is_lucky_user(user):
            return PossibleMiddlewareResultt(something="foo")  # this will get returned no other middleware will run
        elif user.password != input.password:
            return WrongPasswordError("wrong password")  # an exception will get returned
        return  # nothing happens, next middleware or field will run normally

@strawberry.type
class Query
    @strawberry.field(middleware=[MyMiddleware])
    async def my_field(self, input: Input) -> MyQueryResult: ...

Here MyMiddleware.run returns PossibleMiddleWareResult or WrongPasswordError. These types are also part of MyQueryResult union, which allows us to use that middleware on my_field. If types are different, strawberry should raise a runtime exception. MiddlewareResponse here works almost like Optional, by providing possibility that nothing will be returned (in that case, next middleware or field will run, as @dominiquegarmier suggested)

I really like idea of being able to chain middlewares together. However, I'm not sure on how user will define such chains. Using just a list seems a bit dirty to me. But maybe that's me overthinking it because I can't find any real reason to why that would be inconvenient.

Also I've replaced raising ValueError, since I don't like the whole idea of raising an exception. That causes error to go to errors section of request. But as described here, it's better to put at least some of the errors to data section to be able to provide error details to the user. That creates inconsistency, where some of the errors are displayed in one section and other are in another. So I tend to put all of the "expected" errors to data section. Unfortunately, PermissionErrors are placed into errors section and to my knowledge there's no workaround to change that behavior. If middlewares were to be implemented, that could completely replace permissions. That will help to solve that inconsistency.

Overall, middlewares just seem like "better permissions" with some useful features, so in my opinion they would be quite a useful addition.

[dominiquegarmier]

So I thought about this some more. I think it would be nice if the response of such a middleware wouldn't be strictly tied to a single response type. Instead implement it as a dict internally, and allow any return types that will allow Response(**mw_response). If you have a bunch of return types that all have an attribute errors, you could share middleware between them as long as the middleware returns a dict {"errors": ...}.

Here is some code I came up with.

from typing import TypeVar
from typing import Generic
from typing import Type

import strawberry


K = TypeVar('K')


# the generic type indicates to what type
# we can coerce the response to be in _get_resp
class MiddlewareResponse(Generic[K]):
    def __init__(self, **kwargs):
        self.vars = {k: v for k, v in kwargs.items()}

    def _get_resp(self, cls: Type[K]) -> K:
        return cls(**self.vars)


T = TypeVar('T')


class BaseMiddleware(Generic[T]):

    _ret_cls: Type[T]

    def __init__(self, return_cls: Type[T]) -> None:
        self._ret_cls = return_cls

    # this method takes **kwargs so you are flexible with
    # how you want to use args
    # however it should return responses that can be coerced
    # to be of type T, as explained above
    def run(self, **kwargs) -> MiddlewareResponse[T] | None:
        raise NotImplementedError()


class MyMiddleware(BaseMiddleware):
    def run(self, **kwargs) -> MiddlewareResponse | None:
        if True:
            return MiddlewareResponse(foo='bar')  # return early
        elif False:
            raise Exception('foo')  # raise exception
        else:
            return None  # do nothing, run next


# this logic would live in strawberry.field, ...
def middleware_runner(input, field, middleware_list: list[BaseMiddleware]):
    for middleware in middleware_list:
        try:
            mw_resp = middleware.run()
        except Exception as e:  # perhaps logging?
            raise
        if isinstance(mw_resp, MiddlewareResponse):
            return mw_resp._get_resp(middleware._ret_cls)

    return field(input)


@strawberry.input
class Input:
    ...


@strawberry.field
class Response:
    foo: str = 'foo'
    bar: str = 'bar'


# we would now define middleware instances like so
mw_list = [MyMiddleware(Response)]

Im not sure if mypy will understand that MyMiddleware(Response) has type MyMiddleware[Response] and I'm also not sure if there are better ways to use generics here.

as for what you said about Exceptions. Yes I agree you probably shouldn't raise exceptions in middleware, its better to return graphql error types. But that's a larger question for strawberry, how to handle errors.

[dominiquegarmier]

ok, so in that case we would have to implement field-wrappers (decorators?)? if we want to give full control to the user?

there are two problems with that:

1. too many decorators may be hard to read / not sure if it's very pythonic? (I'm not really decided on that point)

   possible solution: we could accept the wrappers as a list as mentioned above and then execute them in order?

2. correctly type annotating decorators is not straightforward / strawberry will have to be able to correctly infer the types

   possible solution: create some decorators from which to inherit from?

[LeKSuS-04]

What if middleware were made to act almost like decorators? Not sure on implementation details, but for user it would look like this:

@strawberry.type
class Query
    my_field: MyType = strawberry.field(my_resolver, middleware=[FirstMiddleware, SecondMiddleware])

... works as ...

class Query:
    my_field: MyType = @strawberry.field(FirstMiddleware.run(SecondMiddleware.run(my_resolver)))

Where MyType encapsulates all possible outcomes.

[dominiquegarmier]

yes exactly, I would suggest that we would offer both apis, decorator and as a list. Since internally it would be built the same way. Similar to how you can use strawberry.field as a decorator and as a method.

[dominiquegarmier]

what im currently struggling with is:

say we have some fields:

def foo() -> int: ...
def bar() -> str: ...

and we would like our middleware to return a generic Response[T, V] such that:

def middleware(func: Callable[..., T]) -> Callable[..., Response[T]]: ...

then middleware(foo) and middleware(bar) should have the corresponding return types Response[int] and Response[str]. But at the moment I'm unable to forward these annotations properly. I would have to look at inspecting and setting the types at runtime. I have no idea how that works.

[dominiquegarmier]

the reason I'm bringing this up is, it would be nice if we could build generic middleware.

I was for example imagining an ErrorHandler middleware which could get instantiated with the response and expected error types mw = ErrorHandler(InnerReponseType, [ErrorOne, ErrorTwo]). Where InnerResponseType, ErrorOne and ErrorTwo are strawberry.type's.

And now with a field def field(...) -> InnerResponseType and middleware as above, this would get turned into having a return type of Response[InnerResponse, Union[ErrorOne, ErrorTwo]] or something like that.

What Response[...] is defined as and how the conversion from field() to Response[...] happens would be left to the user to define.

EDIT:
i just found this:

strawberry/strawberry/types/fields/resolver.py

Line 30 in fae03c5

type_override: Optional[Union[StrawberryType, type]] = None,