Add SchemaExtension.should_await

strawberry/extensions/base_extension.py

@@ -50,6 +50,16 @@
         """Called before and after the execution step."""
         yield None
 
+    def should_await(self, _next: Callable) -> bool:
+        """Whether the extension should await the result from next.
+
+        This relies on the _is_async attribute set by the `SchemaConverter`,
+        normally you'd use `inspect.isawaitable` instead, but that has
+        some performance hits, especially because we know if the resolver
+        is async or not at schema creation time.
+        """
+        return _next._is_async  # type: ignore
+
     def resolve(
         self,
         _next: Callable,

strawberry/schema/schema_converter.py

@@ -757,9 +757,11 @@ async def _async_resolver(
 
         if field.is_async:
             _async_resolver._is_default = not field.base_resolver  # type: ignore
+            _async_resolver._is_async = True  # type: ignore
             return _async_resolver
         else:
             _resolver._is_default = not field.base_resolver  # type: ignore
+            _resolver._is_async = False  # type: ignore
             return _resolver
 
     def from_scalar(self, scalar: Type) -> GraphQLScalarType:

tests/benchmarks/test_execute_with_extensions.py

@@ -21,7 +21,7 @@ def get_results(self) -> AwaitableOrValue[Dict[str, Any]]:
 class ResolveExtension(SchemaExtension):
     async def resolve(self, _next, root, info, *args: Any, **kwargs: Any) -> Any:
         result = _next(root, info, *args, **kwargs)
-        if isawaitable(result):
+        if (hasattr(_next, "_is_async") and _next._is_async) or isawaitable(result):
             result = await result
         return result