Graphene-django: How to let GraphQL read Streamfield

[zxchew2014]

Any ideas how can I solve this problem to let StreamField read on GraphQL?

Below are the code and error show. after i run localhost:8080/graphql

schema.py

import graphene
from graphene_django import DjangoObjectType
from pages.models import Page

class PageType(DjangoObjectType):
    class Meta:
        model = Page

# Query
class Query(graphene.ObjectType):
    pages = graphene.List(PageType)

    def resolve_pages(self, info, **kwargs):
        return Page.objects.all()

models.py

from wagtail.wagtailadmin.edit_handlers import TabbedInterface, ObjectList, StreamFieldPanel
from wagtail.wagtailcore.models import Page as WagtailPage, Orderable
from django.http import JsonResponse
from wagtail.wagtailcore.fields import StreamField
from wagtail.wagtailcore import blocks
from wagtail.wagtailimages.blocks import ImageChooserBlock
from wagtail.wagtaildocs.blocks import DocumentChooserBlock
from wagtail.wagtailsnippets.blocks import SnippetChooserBlock
from wagtail.wagtailsnippets.edit_handlers import SnippetChooserPanel
from wagtail.wagtailembeds.blocks import EmbedBlock
from modelcluster.fields import ParentalKey

from django.db import models


class PageIndex:
    # Parent page / subpage type rules
    parent_page_types = []
    subpage_types = []

    def serve(self, request):
        return JsonResponse({
            'title': self.title,
            'body': self.body,
        })


class CarouselBlock(blocks.StreamBlock):
    image = ImageChooserBlock()
    caption = blocks.TextBlock(blank=True)

    class Meta:
        icon = 'image'


class Page(WagtailPage):
    body = StreamField([
        ('title', blocks.TextBlock(icon="title")),
        ('paragraph', blocks.RichTextBlock(editor='tinymce')),
        ('url', blocks.URLBlock()),
        ('blockquote', blocks.BlockQuoteBlock()),
        ('document', DocumentChooserBlock()),
        ('image', ImageChooserBlock()),
        ('media', EmbedBlock()),
        ('snippet', SnippetChooserBlock(target_model='contents.Content')),
    ], blank=True, null=True)

    content_panels = WagtailPage.content_panels + [
        StreamFieldPanel('body'),
    ]

    edit_handler = TabbedInterface([
        ObjectList(content_panels, heading='Contents'),
        ObjectList(WagtailPage.promote_panels, heading='Promote'),
        ObjectList(WagtailPage.settings_panels, heading='Settings', classname="settings"),
    ])

    def get_url_parts(self, *args, **kwargs):
        super(Page, self).get_url_parts(*args, **kwargs)


# Snippet Model
class PageContentPlacement(Orderable, models.Model):
    page = ParentalKey('pages.Page', related_name='content_placements')
    content = models.ForeignKey('contents.Content', related_name='+')

    class Meta:
        verbose_name = "content placement"
        verbose_name_plural = "content placements"

    panels = [
        SnippetChooserPanel('content'),
    ]

    def __str__(self):
        return self.page.title + " -> " + self.content.description

[patrick91]

@zxchew2014 you can do a custom type for it and return the json :)
Sorry I don't have much time to test it, but it should work, I might have an example somewhere.

Let me know if you can't make it work

[zxchew2014]

@patrick91 i cannot return json in graphql

[zxchew2014]

in my localhost:8080/graphql show

{
  "data": {
    "pages": [
      {
       "id": "3",
        "body": "<div class=\"block-title\">Hello World</div>"
      },
]}}

in my postgreSQL

[
  {
    "type": "title",
    "value": "Hello World",
    "id": "2d06a486-1f8e-4242-b58f-15838aad7ed4"
  }
]

i add few line inside my pages/schema.py

from wagtail.wagtailcore.fields import StreamField
from graphene_django.converter import convert_django_field

@convert_django_field.register(StreamField)
def convert_stream_field(field, registry=None):
    # Customization here
    return graphene.String()

How can i show in my graphql my body show json object instead of they rendering into html tag

@patrick91 any ideas how?

[patrick91]

@zxchew2014 check this comment here: https://github.com/graphql-python/graphene-django/issues/269#issuecomment-362934574

[zxchew2014]

Thanks @patrick91

[zxchew2014]

"body": [ { "type": "call_to_action", "value": { "background_image": null, "title": "Hello World (Pricing)", "description": "", "call_to_action": [ { "type": "page", **"value": 6,** "id": "26b47ccd-f2ce-4ce4-a771-986eacd17eb3" } ] }, "id": "ec808d0a-b95f-489b-b68e-ddfc951b84b0" } ]

type: "page" is a PageChooseBlock under my StreamFields

Any ideas how can i retrieve out as a JSON object instead of just showing the ID?

Expected Output:
"body": [ { "type": "call_to_action", "value": { "background_image": null, "title": "Hello World (Pricing)", "description": "", "call_to_action": [ { "type": "page", "value": **{ **"id" : 6,** "title": "Hello World", "url_path" : "/en/hello-world/" },** "id": "26b47ccd-f2ce-4ce4-a771-986eacd17eb3" } ] }, "id": "ec808d0a-b95f-489b-b68e-ddfc951b84b0" } ]

[zxchew2014]

@patrick91 any ideas how can do it

[fredmontet]

Hi! Did you find an answer?

[osartun]

Hey guys

I'm reviving this conversation for a second.
I used @patrick91's solution and it works to output the contents of a StreamField as JSON:
https://github.com/patrick91/wagtail-ql/blob/963ab22c056d0bc0f6d6955244a03feb475efc7b/backend/graphene_utils/converter.py
The actual value comes from the serialize method:

    def serialize(dt):
        return dt.stream_data

Now, I was super happy at first with this solution because finally I was able to create GraphQL queries without getting errors.
That was until I realized that referenced objects are only included by their ID.

"content": [
  …
  {
    "id": "007c2cfc-3512-4a39-96ba-086d3d9b5300",
    "type": "feature",
    "value": {
      "color": "#08da8a",
      "description": "…",
      "headline": "Activity tracking",
      "screenshot": 4 // <= See here
    }
  },
  {
    "id": "1e0e7d83-115d-427f-a087-f10eca4b09ce",
    "type": "testimonials",
    "value": {
      "headline": "Real results",
      "testimonials": [
        1, // <= And also here
        2
      ],
      "text": "…"
    }
  }
]

When I have a referenced field I'd like to get all the data I want instead of just the ID.
I guess this would be possible with Inline Fragments.

content {
  ... on Image {
    // …
  }
  ... on Testimonial {
    name
    gender
    age
    quote
  }
}

However, this is my first implementation of a GraphQL schema and also my first time working with Wagtail, Django and Python. It's hard for me figuring out how I could do that.
In general I think there's a need for a consistent and flexible solution for StreamFields in wagtail. So, maybe something like this should be added and released as a package on its own.

[osartun]

I created an implementation that works! And I'm posting it here, so that nobody has to struggle with it the way I did.

I'm not sure how Pythonic this solution is though.

First: I wanted to have a Union type for the StreamField so that I can use Inline Fragments in the GraphQL query for each block.
As there are potentially many kinds of StreamFields in a wagtail app with different kinds of blocks I wanted to have a function that generates the Union type the way I want to have it.
So, I created create_stream_field_type it returns a tuple with the graphene schema type and a resolver function.

# utils.py
import graphene

# …

def create_stream_field_type(field_name, **kwargs):
    block_type_handlers = kwargs.copy()

    class StreamFieldType(graphene.Union):
        class Meta:
            types = (GenericStreamBlock, ) + tuple(
                block_type_handlers.values())

    def convert_block(block):
        block_type = block.get('type')
        value = block.get('value')
        if block_type in block_type_handlers:
            handler = block_type_handlers.get(block_type)
            if isinstance(value, dict):
                return handler(value=value, block_type=block_type, **value)
            else:
                return handler(value=value, block_type=block_type)
        else:
            return GenericStreamBlock(value=value, block_type=block_type)

    def resolve_field(self, info):
        field = getattr(self, field_name)
        if not isinstance(field, StreamValue):
            raise Exception(
                f"Field '{field_name}' of {type(self)} not a StreamField")
        return [convert_block(block) for block in field.stream_data]

    return (graphene.List(StreamFieldType), resolve_field)

By default it resolves each block as a GenericStreamBlock. This is its implementation:

# utils.py
import graphene
from graphene.types.generic import GenericScalar

# …

class GenericStreamBlock(graphene.ObjectType):
    block_type = graphene.String()
    value = GenericScalar()

This is how you use it:

# schema.py

class PromoPageNode(DjangoObjectType):
    content, resolve_content = create_stream_field_type('content')
    …

And this is how a query would look like:

{
  allPromoPages(language: "en") {
    content {
      ... on GenericStreamBlock {
        blockType
        value
      }
    }
  }
}

Now, this is cool and all, but it actually doesn't help in any way.

To use the power of inline fragments you need to define a custom type for each block.
Let's start with a ParagraphBlock.

# schema.py
from .utils import GenericStreamBlock, create_stream_field_type

# …

class ParagraphBlock(GenericStreamBlock):
    pass

# …

class PromoPageNode(DjangoObjectType):
    content, resolve_content = create_stream_field_type('content', paragraph=ParagraphBlock)
    # …

All paragraph blocks are now resolved as a ParagraphBlock.

The query could look like this now:

{
  allPromoPages(language: "en") {
    content {   
      ... on ParagraphBlock {
        value
      }
      ... on GenericStreamBlock {
        blockType
        value
      }
    }
  }
}

My paragraph blocks are pretty simple. I have some more complex StructBlocks though and this is where this implementation comes in really handy.

# schema.py
from promo.snippets import Testimonial
from .utils import GenericStreamBlock, create_stream_field_type

# …

class TestimonialNode(DjangoObjectType):
    class Meta:
        model = Testimonial


class FeatureBlock(GenericStreamBlock):
    headline = graphene.String()
    color = graphene.String()
    description = graphene.String()
    screenshot = graphene.ID()


class TestimonialBlock(GenericStreamBlock):
    headline = graphene.String()
    text = graphene.String()
    testimonials = graphene.List(TestimonialNode)

    def resolve_testimonials(self, info):
        testimonial_ids = self.value.get('testimonials')
        return Testimonial.objects.filter(id__in=testimonial_ids)


class PromoPageNode(DjangoObjectType):
    content, resolve_content = create_stream_field_type(
        'content', paragraph=ParagraphBlock, feature=FeatureBlock, testimonials=TestimonialBlock)

Finally this gives me all the power to create a query like this:

{
  allPromoPages(language: "en") {
    content {   
      ... on ParagraphBlock {
        value
      }
      ... on GenericStreamBlock {
        blockType
        value
      }
      ... on FeatureBlock {
        headline
        color
        description
      }
      ... on TestimonialBlock {
        headline
        text
        testimonials {
          id
          name
          gender
          age
        }
      }
    }
  }
}

I hope this helps people who also want to implement a powerful GraphQL solution for StreamFields with Graphene.

[dopeboy]

Thanks @osartun for the detailed write up.

@zxchew2014, did you find @osartun's solution to work for you?

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.