Add support for multipart subscriptions

RELEASE.md

@@ -0,0 +1,18 @@
+Release type: minor
+
+This release adds support for multipart subscriptions in almost all[^1] of our
+http integrations!
+
+[Multipart subcriptions](https://www.apollographql.com/docs/router/executing-operations/subscription-multipart-protocol/)
+are a new protocol from Apollo GraphQL, built on the
+[Incremental Delivery over HTTP spec](https://github.com/graphql/graphql-over-http/blob/main/rfcs/IncrementalDelivery.md),
+which is also used for `@defer` and `@stream`.
+
+The main advantage of this protocol is that when using the Apollo Client
+libraries you don't need to install any additional dependency, but in future
+this feature should make it easier for us to implement `@defer` and `@stream`
+
+Also, this means that you don't need to use Django Channels for subscription,
+since this protocol is based on HTTP we don't need to use websockets.
+
+[^1]: Flask, Chalice and the sync Django integration don't support this.

TWEET.md

@@ -0,0 +1,5 @@
+🆕 Release $version is out! Thanks to $contributor for the PR 👏
+
+Strawberry GraphQL now supports @apollographql's multipart subscriptions! 🎉
+
+Get it here 👉 $release_url

docs/README.md

@@ -12,6 +12,7 @@ title: Strawberry docs
 - [Queries](./general/queries.md)
 - [Mutations](./general/mutations.md)
 - [Subscriptions](./general/subscriptions.md)
+- [Multipart Subscriptions](./general/multipart-subscriptions.md)
 - [Why](./general/why.md)
 - [Breaking changes](./breaking-changes.md)
 - [Upgrading Strawberry](./general/upgrades.md)

docs/general/multipart-subscriptions.md

@@ -0,0 +1,27 @@
+---
+title: Multipart subscriptions
+---
+
+# Multipart subscriptions
+
+Strawberry supports subscription over multipart responses. This is an
+[alternative protocol](https://www.apollographql.com/docs/router/executing-operations/subscription-multipart-protocol/)
+created by [Apollo](https://www.apollographql.com/) to support subscriptions
+over HTTP, and it is supported by default by Apollo Client.
+
+# Support
+
+We support multipart subscriptions out of the box in the following HTTP
+libraries:
+
+- Django (only in the Async view)
+- ASGI
+- Litestar
+- FastAPI
+- AioHTTP
+- Quart
+
+# Usage
+
+Multipart subscriptions are automatically enabled when using Subscription, so no
+additional configuration is required.

docs/integrations/index.md

@@ -0,0 +1,12 @@
+# integrations
+
+WIP:
+
+| name                        | Supports sync | Supports async       | Supports subscriptions via websockets | Supports subscriptions via multipart HTTP | Supports file uploads | Supports batch queries |
+| --------------------------- | ------------- | -------------------- | ------------------------------------- | ----------------------------------------- | --------------------- | ---------------------- |
+| [django](//django.md)       | ✅            | ✅ (with Async view) | ❌ (use Channels for websockets)      | ✅ (From Django 4.2)                      | ✅                    | ❌                     |
+| [starlette](//starlette.md) | ✅            | ✅                   | ✅                                    | ✅                                        | ✅                    | ✅                     |
+| [aiohttp](//aiohttp.md)     | ✅            | ✅                   | ✅                                    | ✅                                        | ✅                    | ✅                     |
+| [flask](//flask.md)         | ✅            | ✅                   | ❌                                    | ❌                                        | ✅                    | ✅                     |
+| [channels](//channels.md)   | ✅            | ✅                   | ✅                                    | ❌                                        | ✅                    | ✅                     |
+| [fastapi](//fastapi.md)     | ✅            | ✅                   | ✅                                    | ✅                                        | ✅                    | ✅                     |

pyproject.toml

@@ -173,6 +173,12 @@ filterwarnings = [
     # ignoring the text instead of the whole warning because we'd
     # get an error when django is not installed
     "ignore:The default value of USE_TZ",
+    "ignore::DeprecationWarning:pydantic_openapi_schema.*",
+    "ignore::DeprecationWarning:graphql.*",
+    "ignore::DeprecationWarning:websockets.*",
+    "ignore::DeprecationWarning:pydantic.*",
+    "ignore::UserWarning:pydantic.*",
+    "ignore::DeprecationWarning:pkg_resources.*",
 ]
 
 [tool.autopub]

strawberry/aiohttp/views.py

@@ -7,10 +7,13 @@
 from typing import (
     TYPE_CHECKING,
     Any,
+    AsyncGenerator,
+    Callable,
     Dict,
     Iterable,
     Mapping,
     Optional,
+    Union,
     cast,
 )
 
@@ -73,11 +76,17 @@ async def get_form_data(self) -> FormData:
 
     @property
     def content_type(self) -> Optional[str]:
-        return self.request.content_type
+        return self.headers.get("content-type")
 
 
 class GraphQLView(
-    AsyncBaseHTTPView[web.Request, web.Response, web.Response, Context, RootValue]
+    AsyncBaseHTTPView[
+        web.Request,
+        Union[web.Response, web.StreamResponse],
+        web.Response,
+        Context,
+        RootValue,
+    ]
 ):
     # Mark the view as coroutine so that AIOHTTP does not confuse it with a deprecated
     # bare handler function.
@@ -180,5 +189,29 @@ def create_response(
 
         return sub_response
 
+    async def create_multipart_response(
+        self,
+        request: web.Request,
+        stream: Callable[[], AsyncGenerator[str, None]],
+        sub_response: web.Response,
+    ) -> web.StreamResponse:
+        response = web.StreamResponse(
+            status=sub_response.status,
+            headers={
+                **sub_response.headers,
+                "Transfer-Encoding": "chunked",
+                "Content-type": "multipart/mixed;boundary=graphql;subscriptionSpec=1.0,application/json",
+            },
+        )
+
+        await response.prepare(request)
+
+        async for data in stream():
+            await response.write(data.encode())
+
+        await response.write_eof()
+
+        return response
+
 
 __all__ = ["GraphQLView"]

strawberry/asgi/__init__.py

@@ -5,6 +5,8 @@
 from typing import (
     TYPE_CHECKING,
     Any,
+    AsyncIterator,
+    Callable,
     Mapping,
     Optional,
     Sequence,
@@ -14,7 +16,12 @@
 
 from starlette import status
 from starlette.requests import Request
-from starlette.responses import HTMLResponse, PlainTextResponse, Response
+from starlette.responses import (
+    HTMLResponse,
+    PlainTextResponse,
+    Response,
+    StreamingResponse,
+)
 from starlette.websockets import WebSocket
 
 from strawberry.asgi.handlers import (
@@ -213,3 +220,19 @@ def create_response(
             response.status_code = sub_response.status_code
 
         return response
+
+    async def create_multipart_response(
+        self,
+        request: Request | WebSocket,
+        stream: Callable[[], AsyncIterator[str]],
+        sub_response: Response,
+    ) -> Response:
+        return StreamingResponse(
+            stream(),
+            status_code=sub_response.status_code,
+            headers={
+                **sub_response.headers,
+                "Transfer-Encoding": "chunked",
+                "Content-type": "multipart/mixed;boundary=graphql;subscriptionSpec=1.0,application/json",
+            },
+        )

strawberry/channels/handlers/http_handler.py

@@ -1,16 +1,21 @@
-"""GraphQLHTTPHandler.
-
-A consumer to provide a graphql endpoint, and optionally graphiql.
-"""
-
 from __future__ import annotations
 
 import dataclasses
 import json
 import warnings
 from functools import cached_property
 from io import BytesIO
-from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional, Union
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    AsyncGenerator,
+    Callable,
+    Dict,
+    Mapping,
+    Optional,
+    Union,
+)
+from typing_extensions import assert_never
 from urllib.parse import parse_qs
 
 from django.conf import settings
@@ -44,6 +49,14 @@ class ChannelsResponse:
     headers: Dict[bytes, bytes] = dataclasses.field(default_factory=dict)
 
 
+@dataclasses.dataclass
+class MultipartChannelsResponse:
+    stream: Callable[[], AsyncGenerator[str, None]]
+    status: int = 200
+    content_type: str = "multipart/mixed;boundary=graphql;subscriptionSpec=1.0"
+    headers: Dict[bytes, bytes] = dataclasses.field(default_factory=dict)
+
+
 @dataclasses.dataclass
 class ChannelsRequest:
     consumer: ChannelsConsumer
@@ -186,16 +199,28 @@ def create_response(
     async def handle(self, body: bytes) -> None:
         request = ChannelsRequest(consumer=self, body=body)
         try:
-            response: ChannelsResponse = await self.run(request)
+            response = await self.run(request)
 
             if b"Content-Type" not in response.headers:
                 response.headers[b"Content-Type"] = response.content_type.encode()
 
-            await self.send_response(
-                response.status,
-                response.content,
-                headers=response.headers,
-            )
+            if isinstance(response, MultipartChannelsResponse):
+                response.headers[b"Transfer-Encoding"] = b"chunked"
+                await self.send_headers(headers=response.headers)
+
+                async for chunk in response.stream():
+                    await self.send_body(chunk.encode("utf-8"), more_body=True)
+
+                await self.send_body(b"", more_body=False)
+
+            elif isinstance(response, ChannelsResponse):
+                await self.send_response(
+                    response.status,
+                    response.content,
+                    headers=response.headers,
+                )
+            else:
+                assert_never(response)
         except HTTPException as e:
             await self.send_response(e.status_code, e.reason.encode())
 
@@ -204,7 +229,7 @@ class GraphQLHTTPConsumer(
     BaseGraphQLHTTPConsumer,
     AsyncBaseHTTPView[
         ChannelsRequest,
-        ChannelsResponse,
+        Union[ChannelsResponse, MultipartChannelsResponse],
         TemporalResponse,
         Context,
         RootValue,
@@ -248,6 +273,16 @@ async def get_context(
     async def get_sub_response(self, request: ChannelsRequest) -> TemporalResponse:
         return TemporalResponse()
 
+    async def create_multipart_response(
+        self,
+        request: ChannelsRequest,
+        stream: Callable[[], AsyncGenerator[str, None]],
+        sub_response: TemporalResponse,
+    ) -> MultipartChannelsResponse:
+        status = sub_response.status_code or 200
+        headers = {k.encode(): v.encode() for k, v in sub_response.headers.items()}
+        return MultipartChannelsResponse(stream=stream, status=status, headers=headers)
+
     async def render_graphql_ide(self, request: ChannelsRequest) -> ChannelsResponse:
         return ChannelsResponse(
             content=self.graphql_ide_html.encode(), content_type="text/html"
@@ -302,7 +337,7 @@ def run(
         request: ChannelsRequest,
         context: Optional[Context] = UNSET,
         root_value: Optional[RootValue] = UNSET,
-    ) -> ChannelsResponse:
+    ) -> ChannelsResponse | MultipartChannelsResponse:
         return super().run(request, context, root_value)