Defer and Stream

Strawberry provides experimental support for GraphQL’s @defer and @stream directives, which enable incremental delivery of response data. These directives allow parts of a GraphQL response to be delivered as they become available, rather than waiting for the entire response to be ready.

Note

This feature is currently experimental. The API and behavior may change in future releases.

Important limitations:

Extensions (most importantly MaskErrors ) are not fully supported yet. Extensions currently only process the initial result and do not handle incremental payloads delivered by @defer and @stream .
This means error masking and other extension functionality will only apply to the initial response, not to deferred or streamed data.
Incremental delivery requires asynchronous execution. Schema.execute_sync and the synchronous integrations (such as the sync Django, Flask and Chalice views) return an error for operations that use @defer or @stream .

Enabling Defer and Stream

To use @defer and @stream directives, you need to enable experimental incremental execution in your schema configuration:

import strawberry
from strawberry.schema.config import StrawberryConfig


@strawberry.type
class Query:
    # Your query fields here
    pass


schema = strawberry.Schema(
    query=Query, config=StrawberryConfig(enable_experimental_incremental_execution=True)
)

Using @defer

The @defer directive allows you to mark parts of a query to be resolved asynchronously. The initial response will include all non-deferred fields, followed by incremental payloads containing the deferred data.

Example

import asyncio
import strawberry


@strawberry.type
class Author:
    id: strawberry.ID
    name: str
    bio: str


@strawberry.type
class Article:
    id: strawberry.ID
    title: str
    content: str

    @strawberry.field
    async def author(self) -> Author:
        # Simulate an expensive operation
        await asyncio.sleep(2)
        return Author(
            id=strawberry.ID("1"),
            name="Jane Doe",
            bio="A passionate writer and developer.",
        )


@strawberry.type
class Query:
    @strawberry.field
    async def article(self, id: strawberry.ID) -> Article:
        return Article(
            id=id,
            title="Introduction to GraphQL Defer",
            content="Learn how to use the @defer directive...",
        )

With this schema, you can query with @defer :

query GetArticle {
  article(id: "123") {
    id
    title
    content
    ... on Article @defer {
      author {
        id
        name
        bio
      }
    }
  }
}

The response will be delivered incrementally:

# Initial payload
{
  "data": {
    "article": {
      "id": "123",
      "title": "Introduction to GraphQL Defer",
      "content": "Learn how to use the @defer directive..."
    }
  },
  "hasNext": true
}

# Subsequent payload
{
  "incremental": [{
    "data": {
      "author": {
        "id": "1",
        "name": "Jane Doe",
        "bio": "A passionate writer and developer."
      }
    },
    "path": ["article"]
  }],
  "hasNext": false
}

Using @stream with strawberry.Streamable

The @stream directive works with list fields and allows items to be delivered as they become available. Strawberry provides a special Streamable type annotation for fields that can be streamed.

Example

import asyncio
import strawberry
from typing import AsyncGenerator


@strawberry.type
class Comment:
    id: strawberry.ID
    content: str
    author_name: str


@strawberry.type
class BlogPost:
    id: strawberry.ID
    title: str

    @strawberry.field
    async def comments(self) -> strawberry.Streamable[Comment]:
        """Stream comments as they are fetched from the database."""
        for i in range(5):
            # Simulate fetching comments from a database
            await asyncio.sleep(0.5)
            yield Comment(
                id=strawberry.ID(f"comment-{i}"),
                content=f"This is comment number {i}",
                author_name=f"User {i}",
            )

Query with @stream :

query GetBlogPost {
  blogPost(id: "456") {
    id
    title
    comments @stream(initialCount: 2) {
      id
      content
      authorName
    }
  }
}

The response will stream the comments:

# Initial payload with first 2 comments
{
  "data": {
    "blogPost": {
      "id": "456",
      "title": "My Blog Post",
      "comments": [
        {
          "id": "comment-0",
          "content": "This is comment number 0",
          "authorName": "User 0"
        },
        {
          "id": "comment-1",
          "content": "This is comment number 1",
          "authorName": "User 1"
        }
      ]
    }
  },
  "hasNext": true
}

# Subsequent payloads for remaining comments
{
  "incremental": [{
    "items": [{
      "id": "comment-2",
      "content": "This is comment number 2",
      "authorName": "User 2"
    }],
    "path": ["blogPost", "comments", 2]
  }],
  "hasNext": true
}
# ... more incremental payloads