Removed resolver_timeout, added support for MiddlewareManager, fixed example.

Author: sergey-komissarov

CHANGELOG.md

@@ -23,6 +23,14 @@ SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 # Changelog
 
+## [1.0.0rc4] - 2023-05-03
+
+- `GraphqlWsConsumer.warn_resolver_timeout` removed.
+- `GraphqlWsConsumer.middleware` accepts an instance of
+  `graphql.MiddlewareManager` or the list of functions. Same as the
+  argument `middleware` of `graphql.execute` method.
+- Fixed broken example.
+
 ## [1.0.0rc3] - 2023-05-02
 
 - Invoke synchronous resolvers in the main thread with eventloop. So

channels_graphql_ws/graphql_ws_consumer.py

@@ -119,18 +119,18 @@ class GraphqlWsConsumer(ch_websocket.AsyncJsonWebsocketConsumer):
     # confirmation is enabled.
     subscription_confirmation_message: Dict[str, Any] = {"data": None, "errors": None}
 
-    # Issue a warning to the log when operation/resolver takes longer
-    # than specified number in seconds. None disables the warning.
+    # Issue a warning to the log when operation takes longer than
+    # specified number in seconds. None disables the warning.
     warn_operation_timeout: Optional[float] = 1
-    warn_resolver_timeout: Optional[float] = 1
 
     # The size of the subscription notification queue. If there are more
     # notifications (for a single subscription) than the given number,
     # then an oldest notification is dropped and a warning is logged.
     subscription_notification_queue_limit: int = 1024
 
     # GraphQL middleware.
-    # List of functions (callables) like the following:
+    # Instance of `graphql.MiddlewareManager` or the list of functions
+    # (callables) like the following:
     # ```python
     # async def my_middleware(next_middleware, root, info, *args, **kwds):
     #     result = next_middleware(root, info, *args, **kwds)
@@ -145,7 +145,7 @@ class GraphqlWsConsumer(ch_websocket.AsyncJsonWebsocketConsumer):
     # - https://graphql-core-3.readthedocs.io/en/latest/diffs.html#custom-middleware
     # Docs about async middlewares are still missing - read the
     # GraphQL-core sources to know more.
-    middleware: Sequence = []
+    middleware: Optional[graphql.Middleware] = None
 
     # Subscription implementation shall return this to tell consumer
     # to suppress subscription notification.
@@ -278,6 +278,15 @@ def __init__(self, *args, **kwargs):
             weakref.WeakValueDictionary()
         )
 
+        # MiddlewareManager maintains internal cache for resolvers
+        # wrapped with middlewares. Using the same manager for all
+        # operations improves performance.
+        self._middleware = None
+        if self.middleware:
+            self._middleware = self.middleware
+            if not isinstance(self._middleware, graphql.MiddlewareManager):
+                self._middleware = graphql.MiddlewareManager(*self._middleware)
+
         super().__init__(*args, **kwargs)
 
     # ---------------------------------------------------------- CONSUMER EVENT HANDLERS
@@ -595,27 +604,6 @@ async def _on_gql_start(self, op_id, payload):
             assert doc_ast is not None
             assert op_ast is not None
 
-            async def unbound_root_middleware(*args, **kwds):
-                """Unbound function for root middleware.
-
-                `graphql.MiddlewareManager` accepts only unbound
-                functions as middleware.
-                """
-                return await self._on_gql_start__root_middleware(
-                    op_id, op_name, *args, **kwds
-                )
-
-            # NOTE: Middlewares order is important, root middleware
-            # should always be the farest from the real resolver (last
-            # in the middleware list). Because we want to calculate
-            # resolver execution time with middlewares included.
-            middlewares = list(self.middleware)
-            if self.warn_resolver_timeout is not None:
-                middlewares.append(unbound_root_middleware)
-            middleware_manager: Optional[graphql.MiddlewareManager] = None
-            if middlewares:
-                middleware_manager = graphql.MiddlewareManager(*middlewares)
-
             # If the operation is subscription.
             if op_ast.operation == graphql.language.ast.OperationType.SUBSCRIPTION:
                 LOG.debug(
@@ -637,7 +625,7 @@ async def unbound_root_middleware(*args, **kwds):
                         op_id,
                         op_name,
                     ),
-                    middleware=middleware_manager,
+                    middleware=self._middleware,
                     execution_context_class=self._SubscriptionExecutionContext,
                 )
 
@@ -715,6 +703,7 @@ async def consume_stream():
                 # equals to `__schema`. This is a more robust way. But
                 # it will eat up more CPU pre each query. For now lets
                 # check only a query name.
+                middleware_manager = self._middleware
                 if op_name == "IntrospectionQuery":
                     # No need to call middlewares for the
                     # IntrospectionQuery. There no real resolvers. Only
@@ -901,82 +890,6 @@ async def map_source_to_response(payload: Any) -> graphql.ExecutionResult:
         # Map every source value to a ExecutionResult value.
         return graphql.MapAsyncIterator(result_or_stream, map_source_to_response)
 
-    async def _on_gql_start__root_middleware(
-        self,
-        operation_id: int,
-        operation_name: str,
-        next_middleware,
-        root,
-        info: graphql.GraphQLResolveInfo,
-        *args,
-        **kwds,
-    ):
-        """Root middleware injected right before resolver invocation.
-
-        This middleware issues a warning if resolver execution time
-        exceeds a limit.
-
-        Since this middleware always comes first in the list of
-        middlewares, it always receives resolver as the first
-        argument instead of another middleware.
-
-        This is a part of START message processing routine so the name
-        prefixed with `_on_gql_start__` to make this explicit.
-
-        Args:
-            resolver: Resolver to "wrap" into this middleware
-            root: Anything. Eventually passed to the resolver.
-            info: Passed to the resolver.
-
-        Returns:
-            Any value: result returned by the resolver.
-            AsyncGenerator: when subscription starts.
-        """
-
-        # Unwrap resolver from functools.partial or other wrappers.
-        real_resolver = self._on_gql_start__unwrap(next_middleware)
-
-        # Start measuring resolver execution time.
-        if self.warn_resolver_timeout is not None:
-            start_time = time.perf_counter()
-
-        # Execute resolver.
-        result = next_middleware(root, info, *args, **kwds)
-        if inspect.isawaitable(result):
-            result = await result
-
-        # Warn about long resolver execution if the time limit exceeds.
-        if self.warn_resolver_timeout is not None:
-            duration = time.perf_counter() - start_time
-            if duration >= self.warn_resolver_timeout:
-                pretty_name = f"{real_resolver.__qualname__}"
-                if hasattr(real_resolver, "__self__"):
-                    pretty_name = f"{real_resolver.__self__.__qualname__}.{pretty_name}"
-                LOG.warning(
-                    "Resolver %s took %.3f seconds (>%.3f)!"
-                    " Operation %s(%s), path: %s.",
-                    pretty_name,
-                    duration,
-                    self.warn_resolver_timeout,
-                    operation_name,
-                    operation_id,
-                    info.path,
-                )
-
-        return result
-
-    def _on_gql_start__unwrap(self, fn: Callable) -> Callable:
-        """Auxiliary method which unwraps given function.
-
-        This is a part of START message processing routine so the name
-        prefixed with `_on_gql_start__` to make this explicit.
-        """
-        if isinstance(fn, functools.partial):
-            fn = self._on_gql_start__unwrap(fn.func)
-        elif hasattr(fn, "__wrapped__"):
-            fn = self._on_gql_start__unwrap(fn.__wrapped__)
-        return fn
-
     async def _on_gql_start__initialize_subscription_stream(
         self,
         operation_id: int,

example/example.py

@@ -25,9 +25,9 @@
 from collections import defaultdict
 from typing import Any, DefaultDict, Dict
 
-import asgiref
 import channels
 import channels.auth
+import channels.db
 import channels.routing
 import django
 import django.contrib.admin
@@ -108,23 +108,24 @@ class Arguments:
         username = graphene.String(required=True)
         password = graphene.String(required=True)
 
-    def mutate(self, info, username, password):
+    async def mutate(self, info, username, password):
         """Login request."""
 
         # Ask Django to authenticate user.
-        user = django.contrib.auth.authenticate(username=username, password=password)
+        user = await channels.db.database_sync_to_async(
+            django.contrib.auth.authenticate
+        )(username=username, password=password)
         if user is None:
             return Login(ok=False)
 
         # Use Channels to login, in other words to put proper data to
         # the session stored in the scope. The
         # `info.context.channels_scope` is a reference to Channel
         # `self.scope` member.
-        asgiref.sync.async_to_sync(channels.auth.login)(
-            info.context.channels_scope, user
-        )
+        await channels.auth.login(info.context.channels_scope, user)
         # Save the session, `channels.auth.login` does not do this.
-        info.context.channels_scope["session"].save()
+        session = info.context.channels_scope["session"]
+        await channels.db.database_sync_to_async(session.save)()
 
         return Login(ok=True)
 
@@ -140,7 +141,7 @@ class Arguments:
         chatroom = graphene.String()
         text = graphene.String()
 
-    def mutate(self, info, chatroom, text):
+    async def mutate(self, info, chatroom, text):
         """Mutation "resolver" - store and broadcast a message."""
 
         # Use the username from the connection scope if authorized.
@@ -151,7 +152,9 @@ def mutate(self, info, chatroom, text):
         chats[chatroom].append({"chatroom": chatroom, "text": text, "sender": username})
 
         # Notify subscribers.
-        OnNewChatMessage.new_chat_message(chatroom=chatroom, text=text, sender=username)
+        await OnNewChatMessage.new_chat_message(
+            chatroom=chatroom, text=text, sender=username
+        )
 
         return SendChatMessage(ok=True)
 
@@ -208,15 +211,15 @@ def publish(self, info, chatroom=None):
         )
 
     @classmethod
-    def new_chat_message(cls, chatroom, text, sender):
+    async def new_chat_message(cls, chatroom, text, sender):
         """Auxiliary function to send subscription notifications.
 
         It is generally a good idea to encapsulate broadcast invocation
         inside auxiliary class methods inside the subscription class.
         That allows to consider a structure of the `payload` as an
         implementation details.
         """
-        cls.broadcast_sync(
+        await cls.broadcast(
             group=chatroom,
             payload={"chatroom": chatroom, "text": text, "sender": sender},
         )

pyproject.toml

@@ -34,7 +34,7 @@ build-backend = "poetry.core.masonry.api"
 # Docs: https://python-poetry.org/docs/
 [tool.poetry]
 name = "django-channels-graphql-ws"
-version = "v1.0.0rc3"
+version = "v1.0.0rc4"
 description = """Django Channels based WebSocket GraphQL server with Graphene-like subscriptions"""
 authors = ["Alexander A. Prokhorov <[email protected]>"]
 homepage = "https://github.com/datadvance/DjangoChannelsGraphqlWs"