Add support for multipart subscriptions (#3076)

* POC for multipart subscriptions

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Fix name

* Progress

* WIP

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Make test pass with django async

* Attempt with FastAPI, type fixes

* Improve code a bit

* Improve naming

* Run subs in execute

* Workaround httpx for now

* Aiohttp support

* ASGI

* Flask attempt

* Support for sanic

* Wip channels

* Fix syntax

* Fix various type issues

* Pragma no cover

* Initial feature table

* Relative urls

* Fix channels issue

* Handle heartbeat

* Remove type ignore

* Add blank release file

* Wrap response in payload

* Update integrations

* Improve how we check for content types

* Run channels tests, even if they are broken

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Fix lint

* Fix import

* await cancelled tasks

* Some refactoring

* Remove stale comment

* update type

* Pass request down to creat multipart response

* Fix tests with lowercase headers

* Get support

* Some release notes

* Litestar support

* Add docs

* Remove subresponse

* Fix check

* Fix

* Fix

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Lint

* Sanic fix

* Fixes

* Update release file and add tweet file

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Fix tweet

* Update tweet

* Sub response support

* Update docs

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

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)