PK!ewagtail_graphql/__init__.py __version__ = '0.1.1' PK!jwagtail_graphql/actions.py# python import string from typing import Type, Set # django from django.utils.text import capfirst # graphene import graphene from graphene.types.generic import GenericScalar # graphene_django from graphene_django import DjangoObjectType # wagtail from wagtail.core.fields import StreamField from wagtail.core.models import Page as wagtailPage # wagtail forms from wagtail.contrib.forms.models import AbstractForm # wagtail settings from wagtail.contrib.settings.models import BaseSetting # app from .registry import registry from .permissions import with_page_permissions from .types.settings import settings_registry from .settings import url_prefix_for_site # app types from .types import ( PageInterface, Settings, FormError, FormField, ) def _add_form(cls: Type[AbstractForm], node: str, dict_params: dict) -> Type[graphene.Mutation]: registry.page_prefetch_fields.add(cls.__name__.lower()) dict_params['Meta'].interfaces = (PageInterface,) dict_params['form_fields'] = graphene.List(FormField) def form_fields(self, _info): return list(FormField(name=field_.clean_name, field_type=field_.field_type) for field_ in self.form_fields.all()) dict_params['resolve_form_fields'] = form_fields tp = type(node, (DjangoObjectType,), dict_params) registry.pages[cls] = tp args = type("Arguments", (), {'values': GenericScalar(), "url": graphene.String(required=True)}) _node = node def mutate(_self, info, url, values): url_prefix = url_prefix_for_site(info) query = wagtailPage.objects.filter(url_path=url_prefix + url.rstrip('/') + '/') instance = with_page_permissions( info.context, query.specific() ).live().first() user = info.context.user form = instance.get_form(values, None, page=instance, user=user) if form.is_valid(): # form_submission instance.process_form_submission(form) return registry.forms[_node](result="OK") else: return registry.forms[_node](result="FAIL", errors=[FormError(*err) for err in form.errors.items()]) dict_params = { "Arguments": args, "mutate": mutate, "result": graphene.String(), "errors": graphene.List(FormError), } tp = type(node + "Mutation", (graphene.Mutation,), dict_params) # type: Type[Mutation] registry.forms[node] = tp return tp def _add_page(cls: Type[wagtailPage], node: str, dict_params: dict) -> Type[DjangoObjectType]: registry.page_prefetch_fields.add(cls.__name__.lower()) dict_params['Meta'].interfaces = (PageInterface,) tp = type(node, (DjangoObjectType,), dict_params) registry.pages[cls] = tp return tp # type: Type[DjangoObjectType] def _add_setting(cls: Type[BaseSetting], node: str, dict_params: dict) -> Type[DjangoObjectType]: dict_params['Meta'].interfaces = (Settings,) tp = type(node, (DjangoObjectType,), dict_params) # type: Type[DjangoObjectType] registry.settings[capfirst(cls._meta.verbose_name)] = cls return tp def _add_snippet(cls: type, node: str, dict_params: dict) -> Type[DjangoObjectType]: tp = type(node, (DjangoObjectType,), dict_params) # type: Type[DjangoObjectType] registry.snippets[cls] = tp registry.snippets_by_name[node] = tp return tp def _add_django_model(_cls: type, node: str, dict_params: dict) -> Type[DjangoObjectType]: tp = type(node, (DjangoObjectType,), dict_params) # type: Type[DjangoObjectType] registry.django[node] = tp return tp def _add_streamfields(cls: type, node: str, dict_params: dict, app: str, prefix: str) -> None: from .types.streamfield import ( block_handler, stream_field_handler, ) for field in cls._meta.fields: if isinstance(field, StreamField): field_name = field.name stream_field_name = f"{node}{string.capwords(field_name, sep='_').replace('_', '')}" blocks = field.stream_block.child_blocks handlers = dict( (name, block_handler(block, app, prefix)) for name, block in blocks.items() ) f, resolve = stream_field_handler( stream_field_name, field_name, handlers ) dict_params.update({ field.name: f, "resolve_" + field.name: resolve }) def _register_model(registered: Set[type], cls: type, snippet: bool, app: str, prefix: str) -> None: if cls in registered: return prefix = prefix.format(app=string.capwords(app), cls=cls.__name__) node = prefix + cls.__name__ # dict parameters to create GraphQL type class Meta: model = cls interfaces = tuple() # type: tuple dict_params = {'Meta': Meta} # add streamfield handlers _add_streamfields(cls, node, dict_params, app, prefix) if snippet: _add_snippet(cls, node, dict_params) elif issubclass(cls, AbstractForm): _add_form(cls, node, dict_params) elif issubclass(cls, wagtailPage): _add_page(cls, node, dict_params) elif issubclass(cls, BaseSetting): _add_setting(cls, node, dict_params) else: # Django Model _add_django_model(cls, node, dict_params) registered.add(cls) def add_app(app: str, prefix: str = '{app}') -> None: from django.contrib.contenttypes.models import ContentType from wagtail.snippets.models import get_snippet_models snippets = get_snippet_models() models = [mdl.model_class() for mdl in ContentType.objects.filter(app_label=app).all()] to_register = [x for x in snippets + models if x is not None] registered = set() for cls in to_register: _register_model(registered, cls, cls in snippets, app, prefix) def add_apps_with_settings(settings: dict) -> None: if settings_registry: add_app('wagtail.contrib.settings') apps = settings.get('APPS', []) for app in apps: prefixes = settings.get('PREFIX', {}) if isinstance(prefixes, str): prefix = prefixes else: prefix = prefixes.get(app, '{app}') add_app(app, prefix=prefix) if not apps: import logging logging.warning("No APPS specified for wagtail_graphql") def add_apps() -> None: from .settings import SETTINGS add_apps_with_settings(SETTINGS) PK!I,]]wagtail_graphql/apps.pyfrom django.apps import AppConfig class ApiConfig(AppConfig): name = 'wagtail_graphql' PK! wagtail_graphql/permissions.py# python from typing import Any, Union # django from django.db.models import Q from django.contrib.auth.models import AnonymousUser # wagtail from wagtail.core.query import PageQuerySet from wagtail.core.models import PageViewRestriction, CollectionViewRestriction from wagtail.images.models import ImageQuerySet from wagtail.documents.models import DocumentQuerySet def with_page_permissions(request: Any, queryset: PageQuerySet) -> PageQuerySet: user = request.user # Filter by site if request.site: queryset = queryset.descendant_of(request.site.root_page, inclusive=True) else: # No sites configured return queryset.none() # Get live pages that are public and check groups and login permissions if user == AnonymousUser: queryset = queryset.public() elif user.is_superuser: pass else: current_user_groups = user.groups.all() q = Q() for restriction in PageViewRestriction.objects.all(): if (restriction.restriction_type == PageViewRestriction.PASSWORD) or \ (restriction.restriction_type == PageViewRestriction.LOGIN and not user.is_authenticated) or \ (restriction.restriction_type == PageViewRestriction.GROUPS and not any(group in current_user_groups for group in restriction.groups.all()) ): q = ~queryset.descendant_of_q(restriction.page, inclusive=True) queryset = queryset.filter(q).live() return queryset CollectionQSType = Union[ImageQuerySet, DocumentQuerySet] def with_collection_permissions(request: Any, queryset: CollectionQSType) -> CollectionQSType: user = request.user # Get live pages that are public and check groups and login permissions if user == AnonymousUser: queryset = queryset.public() elif user.is_superuser: pass else: current_user_groups = user.groups.all() q = Q() for restriction in CollectionViewRestriction.objects.all(): if (restriction.restriction_type == CollectionViewRestriction.PASSWORD) or \ (restriction.restriction_type == CollectionViewRestriction.LOGIN and not user.is_authenticated) or \ (restriction.restriction_type == CollectionViewRestriction.GROUPS and not any(group in current_user_groups for group in restriction.groups.all()) ): q &= ~Q(collection=restriction.collection) # q &= ~queryset.filter(collection) descendant_of_q(restriction.page, inclusive=True) queryset = queryset.filter(q) return queryset PK!ݗ,wagtail_graphql/registry.py class RegistryItem(dict): @property def types(self) -> tuple: return tuple(self.values()) class Registry: _django = RegistryItem() _forms = RegistryItem() _pages = RegistryItem() _settings = RegistryItem() _snippets = RegistryItem() _snippets_by_name = RegistryItem() _streamfield_blocks = RegistryItem() _streamfield_scalar_blocks = RegistryItem() _page_prefetch = { 'content_type', 'owner', 'live_revision', 'page_ptr' } @property def blocks(self) -> RegistryItem: return self._streamfield_blocks @property def scalar_blocks(self) -> RegistryItem: return self._streamfield_scalar_blocks @property def django(self) -> RegistryItem: return self._django @property def forms(self) -> RegistryItem: return self._forms @property def pages(self) -> RegistryItem: return self._pages @property def settings(self) -> RegistryItem: return self._settings @property def snippets(self) -> RegistryItem: return self._snippets @property def snippets_by_name(self) -> RegistryItem: return self._snippets_by_name @property def rsnippets(self) -> RegistryItem: return RegistryItem((v, k) for k, v in self._snippets.items()) @property def page_prefetch_fields(self) -> set: return self._page_prefetch @property def models(self) -> dict: models: dict = {} models.update(self.pages) models.update(self.snippets) models.update(self.forms) models.update(self.django) models.update(self.settings) models.update((k, v) for k, v in self.blocks.items() if not isinstance(v, tuple)) return models @property def rmodels(self) -> dict: return dict((v, k) for k, v in self.models.items()) registry = Registry() PK!3Aߩwagtail_graphql/schema.py# django from django.utils.text import camel_case_to_spaces # graphql from graphql import ResolveInfo # graphene import graphene # graphene_django from graphene_django.converter import String # app from .registry import registry from .actions import add_apps # mixins from .types import ( DocumentQueryMixin, ImageQueryMixin, InfoQueryMixin, MenusQueryMixin, PagesQueryMixin, SettingsQueryMixin, SnippetsQueryMixin, ) # api version GRAPHQL_API_FORMAT = (0, 1, 1) # add all the apps from the settings add_apps() class Query(graphene.ObjectType, DocumentQueryMixin, ImageQueryMixin, InfoQueryMixin, MenusQueryMixin, PagesQueryMixin, SettingsQueryMixin, SnippetsQueryMixin, ): # API Version format = graphene.Field(String) def resolve_format(self, _info: ResolveInfo): return '%d.%d.%d' % GRAPHQL_API_FORMAT def mutation_parameters() -> dict: dict_params = {"format": graphene.Field(String)} dict_params.update((camel_case_to_spaces(n).replace(' ', '_'), mut.Field()) for n, mut in registry.forms.items()) return dict_params Mutations = type("Mutation", (graphene.ObjectType,), mutation_parameters() ) schema = graphene.Schema( query=Query, mutation=Mutations, types=list(registry.models.values()) ) PK!wagtail_graphql/settings.py# django from django.conf import settings # graphql from graphql import ResolveInfo if hasattr(settings, 'GRAPHQL_API'): SETTINGS = settings.GRAPHQL_API else: SETTINGS = {} URL_PREFIX = SETTINGS.get('URL_PREFIX', {}) def url_prefix_for_site(info: ResolveInfo): hostname = info.context.site.hostname return URL_PREFIX.get( hostname, info.context.site.root_page.url_path.rstrip('/') ) PK!!wagtail_graphql/types/__init__.pyfrom .core import ( PageInterface, PageLink, Site, User, # mixins InfoQueryMixin, PagesQueryMixin, ) from .documents import ( Document, # mixins DocumentQueryMixin, ) from .forms import FormField, FormError from .images import ( Image, # mixins ImageQueryMixin, ) from .settings import Settings, SettingsQueryMixin from .snippets import SnippetsQueryMixin __all__ = [ # core 'PageInterface', 'PageLink', 'Settings', 'Site', 'User', # documents 'Document', # forms 'FormError', 'FormField', # images 'Image', # mixins 'DocumentQueryMixin', 'ImageQueryMixin', 'InfoQueryMixin', 'MenusQueryMixin', 'PagesQueryMixin', 'SettingsQueryMixin', 'SnippetsQueryMixin', ] # menus try: from .menus import MenusQueryMixin, Menu, MenuItem, SecondaryMenu, SecondaryMenuItem # noqa: F401 __all__.extend([ # menus 'Menu', 'MenuItem', 'SecondaryMenu', 'SecondaryMenuItem', ]) HAS_WAGTAILMENUS = True except ImportError: # type: ignore class _MenusQueryMixin: pass MenusQueryMixin = _MenusQueryMixin HAS_WAGTAILMENUS = False PK!Qwagtail_graphql/types/auth.py# django from django.contrib.auth.models import User as wagtailUser, AnonymousUser # graphene import graphene from graphql.execution.base import ResolveInfo # app types from .core import User class AuthQueryMixin: # User information user = graphene.Field(User) def resolve_user(self, info: ResolveInfo): user = info.context.user if isinstance(user, AnonymousUser): return wagtailUser(id='-1', username='anonymous') return user PK!wagtail_graphql/types/core.py# typings from typing import cast # django from django.contrib.auth.models import User as wagtailUser from django.contrib.contenttypes.models import ContentType # graphql from graphql.execution.base import ResolveInfo from graphql.language.ast import InlineFragment # graphene import graphene # graphene_django from graphene_django import DjangoObjectType from graphene_django.converter import convert_django_field, String, List # wagtail from wagtail.core.models import Page as wagtailPage, Site as wagtailSite from taggit.managers import TaggableManager from wagtail.core.utils import camelcase_to_underscore # app from ..settings import url_prefix_for_site from ..registry import registry from ..permissions import with_page_permissions class User(DjangoObjectType): class Meta: model = wagtailUser exclude_fields = ['password'] class Site(DjangoObjectType): class Meta: model = wagtailSite class PageInterface(graphene.Interface): id = graphene.Int(required=True) title = graphene.String(required=True) url_path = graphene.String() content_type = graphene.String() slug = graphene.String(required=True) path = graphene.String() depth = graphene.Int() seoTitle = graphene.String() numchild = graphene.Int() def resolve_content_type(self, _info: ResolveInfo): self.content_type = cast(ContentType, self.content_type) return self.content_type.app_label + '.' + self.content_type.model_class().__name__ @classmethod def resolve_type(cls, instance, info: ResolveInfo) -> 'PageInterface': if isinstance(instance, int): return registry.pages[type(wagtailPage.objects.filter(id=instance).specific().first())] try: model = registry.pages[instance.content_type.model_class()] except KeyError: raise ValueError("Model %s is not a registered GraphQL type" % instance.content_type.model_class()) return model def resolve_url_path(self, info: ResolveInfo) -> str: self.url_path = cast(str, self.url_path) url_prefix = url_prefix_for_site(info) url = self.url_path if not self.url_path.startswith(url_prefix) else self.url_path[len(url_prefix):] return url.rstrip('/') class PageLink(DjangoObjectType): class Meta: model = wagtailPage interfaces = (PageInterface, ) def resolve_url_path(self: PageInterface, info: ResolveInfo) -> str: url_prefix = url_prefix_for_site(info) url = self.url_path if not self.url_path.startswith(url_prefix) else self.url_path[len(url_prefix):] return url.rstrip('/') @convert_django_field.register(TaggableManager) def convert_field_to_string(field, _registry=None): return List(String, description=field.help_text, required=not field.null) class PagesQueryMixin: if registry.pages: class _Page(graphene.types.union.Union): class Meta: types = registry.pages.types Page = _Page else: Page = PageInterface pages = graphene.List(PageInterface, parent=graphene.Int()) page = graphene.Field(PageInterface, id=graphene.Int(), url=graphene.String() ) def resolve_pages(self, info: ResolveInfo, parent: int = None): query = wagtailPage.objects # prefetch specific type pages selections = set(camelcase_to_underscore(f.name.value) for f in info.field_asts[0].selection_set.selections if not isinstance(f, InlineFragment)) for pf in registry.page_prefetch_fields.intersection(selections): query = query.select_related(pf) if parent is not None: parent_page = wagtailPage.objects.filter(id=parent).first() if parent_page is None: raise ValueError(f'Page id={parent} not found.') query = query.child_of(parent_page) return with_page_permissions( info.context, query.specific() ).live().order_by('path').all() def resolve_page(self, info: ResolveInfo, id: int = None, url: str = None): query = wagtailPage.objects if id is not None: query = query.filter(id=id) elif url is not None: url_prefix = url_prefix_for_site(info) query = query.filter(url_path=url_prefix + url.rstrip('/') + '/') else: raise ValueError("One of 'id' or 'url' must be specified") page = with_page_permissions( info.context, query.select_related('content_type').specific() ).live().first() if page is None: return None return page # Show in Menu show_in_menus = graphene.List(PageLink) def resolve_show_in_menus(self, info: ResolveInfo): return with_page_permissions( info.context, wagtailPage.objects.filter(show_in_menus=True) ).live().order_by('path') class InfoQueryMixin: # Root root = graphene.Field(Site) def resolve_root(self, info: ResolveInfo): user = info.context.user if user.is_superuser: return info.context.site else: return None PK!}PP"wagtail_graphql/types/documents.py# graphql from graphql.execution.base import ResolveInfo # graphene import graphene # graphene_django from graphene_django import DjangoObjectType # graphene_django_optimizer import graphene_django_optimizer as gql_optimizer # wagtail documents from wagtail.documents.models import Document as wagtailDocument # app from ..permissions import with_collection_permissions class Document(DjangoObjectType): class Meta: model = wagtailDocument url = graphene.String() filename = graphene.String() file_extension = graphene.String() def resolve_tags(self: wagtailDocument, _info: ResolveInfo): return self.tags.all() class DocumentQueryMixin: documents = graphene.List(Document) document = graphene.Field(Document, id=graphene.Int(required=True)) def resolve_documents(self, info: ResolveInfo): return with_collection_permissions( info.context, gql_optimizer.query( wagtailDocument.objects.all(), info ) ) def resolve_document(self, info: ResolveInfo, id: int): doc = with_collection_permissions( info.context, gql_optimizer.query( wagtailDocument.objects.filter(id=id), info ) ).first() return doc PK!+99wagtail_graphql/types/forms.py# graphene import graphene # graphene_django from graphene_django.converter import String class FormField(graphene.ObjectType): name = graphene.Field(String) field_type = graphene.Field(String) class FormError(graphene.ObjectType): name = graphene.Field(String) errors = graphene.List(String) PK!\  wagtail_graphql/types/images.py# graphql from graphql.execution.base import ResolveInfo # django from django.urls import reverse # graphene import graphene # graphene_django from graphene_django import DjangoObjectType from graphene_django.converter import convert_django_field # graphene_django_optimizer import graphene_django_optimizer as gql_optimizer # wagtail images from wagtail.images.models import Image as wagtailImage from wagtail.images.views.serve import generate_signature # app from ..permissions import with_collection_permissions @convert_django_field.register(wagtailImage) def convert_image(field, _registry=None): return Image(description=field.help_text, required=not field.null) class Rect(graphene.ObjectType): left = graphene.Int() top = graphene.Int() right = graphene.Int() bottom = graphene.Int() x = graphene.Int() y = graphene.Int() height = graphene.Int() width = graphene.Int() class Image(DjangoObjectType): class Meta: model = wagtailImage exclude_fields = [ 'focal_point_x', 'focal_point_y', 'focal_point_width', 'focal_point_height', ] has_focal_point = graphene.Boolean() focal_point = graphene.Field(Rect) url = graphene.String(rendition=graphene.String()) url_link = graphene.String(rendition=graphene.String()) def resolve_has_focal_point(self: wagtailImage, _info: ResolveInfo): return self.has_focal_point() def resolve_focal_point(self: wagtailImage, _info: ResolveInfo): return self.get_focal_point() def resolve_tags(self: wagtailImage, _info: ResolveInfo): return self.tags.all() def resolve_url(self: wagtailImage, _info: ResolveInfo, rendition: str = None): if not rendition: if not self.has_focal_point(): rendition = "original" else: fp = self.get_focal_point() rendition = 'fill-%dx%d-c100' % (fp.width, fp.height) return generate_image_url(self, rendition) def resolve_url_link(self: wagtailImage, _info: ResolveInfo, rendition: str = None): if not rendition: if not self.has_focal_point(): rendition = "original" else: fp = self.get_focal_point() rendition = 'fill-%dx%d-c100' % (fp.width, fp.height) return self.get_rendition(rendition).url def generate_image_url(image: wagtailImage, filter_spec: str) -> str: signature = generate_signature(image.pk, filter_spec) url = reverse('wagtailimages_serve', args=(signature, image.pk, filter_spec)) return url class ImageQueryMixin: images = graphene.List(Image) image = graphene.Field(Image, id=graphene.Int(required=True)) def resolve_images(self, info: ResolveInfo): return with_collection_permissions( info.context, gql_optimizer.query( wagtailImage.objects.all(), info ) ) def resolve_image(self, info: ResolveInfo, id: int): image = with_collection_permissions( info.context, gql_optimizer.query( wagtailImage.objects.filter(id=id), info ) ).first() return image PK!D~wagtail_graphql/types/menus.py# python from typing import List # graphql from graphql import ResolveInfo # graphene_django import graphene from graphene_django import DjangoObjectType # wagtailmenus from wagtailmenus.models import FlatMenu, FlatMenuItem, MainMenu, MainMenuItem class MenuItem(DjangoObjectType): class Meta: model = MainMenuItem class Menu(DjangoObjectType): class Meta: model = MainMenu only_fields = ['max_levels', 'menu_items'] class SecondaryMenuItem(DjangoObjectType): class Meta: model = FlatMenuItem class SecondaryMenu(DjangoObjectType): class Meta: model = FlatMenu only_fields = ['title', 'handle', 'heading', 'max_levels', 'menu_items'] class MenusQueryMixin: main_menu = graphene.List(Menu) secondary_menu = graphene.Field(SecondaryMenu, handle=graphene.String(required=True)) secondary_menus = graphene.List(SecondaryMenu) def resolve_main_menu(self, _info: ResolveInfo) -> List[MainMenu]: return MainMenu.objects.all() def resolve_secondary_menus(self, _info: ResolveInfo) -> List[FlatMenu]: return FlatMenu.objects.all() def resolve_secondary_menu(self, _info, handle: ResolveInfo) -> FlatMenu: return FlatMenu.objects.filter(handle=handle).first() PK!jj!wagtail_graphql/types/settings.py# graphql from graphql.execution.base import ResolveInfo # graphene import graphene # graphene_django from graphene_django.converter import String # wagtail settings try: from wagtail.contrib.settings.registry import registry as settings_registry except ImportError: settings_registry = None # app from ..registry import registry class Settings(graphene.Interface): __typename = graphene.Field(String) class SettingsQueryMixin: if settings_registry: settings = graphene.Field(Settings, name=graphene.String(required=True)) def resolve_settings(self, _info: ResolveInfo, name): try: result = registry.settings[name].objects.first() except KeyError: raise ValueError(f"Settings '{name}' not found.") return result else: pass PK!8ش!wagtail_graphql/types/snippets.py# django from django.db import models # graphql from graphql.execution.base import ResolveInfo # graphene import graphene # app from ..registry import registry class SnippetsQueryMixin: if registry.snippets: class Snippet(graphene.types.union.Union): class Meta: types = registry.snippets.types snippets = graphene.List(Snippet, typename=graphene.String(required=True)) def resolve_snippets(self, _info: ResolveInfo, typename: str) -> models.Model: node = registry.snippets_by_name[typename] cls = node._meta.model return cls.objects.all() else: pass PK!rvPy""$wagtail_graphql/types/streamfield.py# python from typing import Tuple, Callable # graphql from graphql.execution.base import ResolveInfo # graphene import graphene from graphene.types.generic import GenericScalar from graphene.types import Scalar # graphene_django from graphene_django.converter import convert_django_field, List # wagtail import wagtail.core.blocks import wagtail.images.blocks import wagtail.snippets.blocks from wagtail.core.blocks import Block, ListBlock, StructBlock from wagtail.core.fields import StreamField # app from ..registry import registry # app types from .core import PageInterface, wagtailPage from .images import Image, wagtailImage # types StreamFieldHandlerType = Tuple[graphene.List, Callable[[StreamField, ResolveInfo], list]] @convert_django_field.register(StreamField) def convert_stream_field(field, _registry=None): return Scalar(description=field.help_text, required=not field.null) def _scalar_block(graphene_type): tp = registry.scalar_blocks.get(graphene_type) if not tp: node = '%sBlock' % graphene_type tp = type(node, (graphene.ObjectType,), { 'value': graphene.Field(graphene_type), 'field': graphene.Field(graphene.String), }) registry.scalar_blocks[graphene_type] = tp return tp, lambda x: tp def _resolve_scalar(key, type_): def resolve(self, _info: ResolveInfo): return type_(value=self, field=key) return resolve def stream_field_handler(stream_field_name: str, field_name: str, block_type_handlers: dict) -> StreamFieldHandlerType: for k, t in block_type_handlers.items(): # Unions must reference NamedTypes, so for scalar types we need to create a new type to encapsulate the scalar if not isinstance(t, tuple) and issubclass(t, Scalar): typ, typ_fn = _scalar_block(t) block_type_handlers[k] = typ_fn, _resolve_scalar(k, typ) types_ = list(block_type_handlers.values()) for i, t in enumerate(types_): if isinstance(t, tuple): types_[i] = t[0](None) class Meta: types = tuple(set(types_)) def resolve_type(self, _info: ResolveInfo): return self.__class__ stream_field_type = type( stream_field_name + "Type", (graphene.Union, ), { 'Meta': Meta, 'resolve_type': resolve_type } ) def convert_block(block, info: ResolveInfo): 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(handler, tuple): tp, resolver = handler return resolver(value, info) else: if isinstance(value, dict): return handler(**value) else: raise NotImplementedError() else: raise NotImplementedError() def resolve_field(self, info: ResolveInfo): field = getattr(self, field_name) return [convert_block(block, info) for block in field.stream_data] return graphene.List(stream_field_type), resolve_field def _is_compound_block(block): return isinstance(block, StructBlock) def _is_list_block(block): return isinstance(block, ListBlock) def _is_custom_type(block): return hasattr(block, "__graphql_type__") def _add_handler_resolves(dict_params, block): to_add = {} for k, v in dict_params.items(): if isinstance(v, tuple): if isinstance(v[0], (List, graphene.Field)): val = v[0] else: val = v[0](block) to_add['resolve_' + k] = v[1] elif not issubclass(v, Scalar): val = v to_add['resolve_' + k] = _resolve else: val = v dict_params[k] = graphene.Field(val) dict_params.update(to_add) def _class_full_name(cls): return cls.__module__ + "." + cls.__qualname__ def block_handler(block: Block, app, prefix=''): cls = block.__class__ handler = registry.blocks.get(cls) if handler is None: if _is_custom_type(block): target_block_type = block.__graphql_type__() this_handler = block_handler(target_block_type, app, prefix) if isinstance(this_handler, tuple): raise NotImplementedError() if hasattr(block, '__graphql_resolve__'): resolver = _resolve_custom(block, this_handler) elif issubclass(target_block_type, Scalar): resolver = _resolve_generic_scalar else: raise TypeError("Non Scalar custom types need an explicit __graphql_resolve__ method.") handler = (lambda x: this_handler, resolver) elif _is_compound_block(block): node = prefix + cls.__name__ dict_params = dict( (n, block_handler(block_type, app, prefix)) for n, block_type in block.child_blocks.items() ) _add_handler_resolves(dict_params, block) tp = type(node, (graphene.ObjectType,), dict_params) handler = tp registry.blocks[cls] = handler elif _is_list_block(block): this_handler = registry.blocks.get(block.child_block.__class__) if this_handler is None: this_handler = block_handler(block.child_block, app, prefix) if isinstance(this_handler, tuple): this_handler = this_handler[0](block.child_block) handler = List(this_handler), _resolve_list else: handler = List(this_handler), _resolve_simple_list else: handler = GenericScalar return handler def _snippet_handler(block): tp = registry.snippets[block.target_model] return tp def _resolve_snippet(self, info: ResolveInfo): if self is None: return None id_ = getattr(self, info.field_name) cls = info.return_type.graphene_type._meta.model obj = cls.objects.filter(id=id_).first() return obj def _resolve_image(self, info: ResolveInfo): if self is None: return None id_ = getattr(self, info.field_name) return wagtailImage.objects.filter(id=id_).first() def _resolve_page(self, info: ResolveInfo): if self is None: return None id_ = self if isinstance(self, int) else getattr(self, info.field_name) return wagtailPage.objects.filter(id=id_).specific().first() def _resolve(self, info: ResolveInfo): data = getattr(self, info.field_name) cls = info.return_type return cls.graphene_type(**data) def _resolve_custom(block, hdl): def _inner(self, info: ResolveInfo): cls = info.return_type if isinstance(self, dict): data = self else: data = getattr(self, info.field_name) value = block.__graphql_resolve__(data, info) if hasattr(cls, "serialize"): return cls.serialize(value) return hdl(**value) return _inner def _resolve_generic_scalar(self, info: ResolveInfo): data = getattr(self, info.field_name) cls = info.return_type return cls.serialize(data) def _resolve_simple_list(self, info: ResolveInfo): if self is None: return None data = getattr(self, info.field_name) cls = info.return_type.of_type.graphene_type if issubclass(cls, Scalar): return list(d for d in data) return list(cls(**d) for d in data) def _resolve_list(self, info: ResolveInfo): if self is None: return None ids = getattr(self, info.field_name) cls = info.return_type.of_type.graphene_type._meta.model objs = dict((x.id, x) for x in cls.objects.filter(id__in=ids).all()) return list(objs.get(i) for i in ids) registry.blocks.update({ # choosers wagtail.images.blocks.ImageChooserBlock: (lambda x: Image, _resolve_image), wagtail.core.blocks.PageChooserBlock: (lambda x: PageInterface, _resolve_page), wagtail.snippets.blocks.SnippetChooserBlock: (_snippet_handler, _resolve_snippet), # standard fields wagtail.core.blocks.CharBlock: graphene.types.String, wagtail.core.blocks.URLBlock: graphene.types.String, wagtail.core.blocks.DateBlock: graphene.types.Date, wagtail.core.blocks.DateTimeBlock: graphene.types.DateTime, wagtail.core.blocks.BooleanBlock: graphene.types.Boolean, wagtail.core.blocks.IntegerBlock: graphene.types.Int, wagtail.core.blocks.FloatBlock: graphene.types.Float, wagtail.core.blocks.DecimalBlock: graphene.types.String, wagtail.core.blocks.TextBlock: graphene.types.String, }) PK!X//'wagtail_graphql-0.1.2.dist-info/LICENSEMIT License Copyright (c) 2018 Tiago Requeijo Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!H|SU%wagtail_graphql-0.1.2.dist-info/WHEELA н#J;/"d&F[xjxڠ5 G7#hi"h4?h>0k:baPn^PK!H"Gk(wagtail_graphql-0.1.2.dist-info/METADATAWr6}W`M;jۉeƓ LgP7'=Lpg.[Sh#UeAt es>\U1"A|]Ez%UŬbV&<) UuB/oH+O32Ng*mVD:۱ d"JuX̕N2VkJU{͔kɧ]T8n]fe8wL E´ WGˣu΍)!ɺ]UJ;3n^o BE!)rZ迣/c;"{y*ilW6C=u|{!{*+ Ф?^A3[? ! 1{^?>ĶtQJJ/Z&߉0H@תE6Z8+fQ^n3iM8~:97d`%JO!h!Gf(F &fי`jL`;u4>c'E 4O,6KBĐ@S_ 2`,xmKVq H|9и.B4V׉5*/Ž 7(Bb7ºqH3QnF#<*Y- (ꡲ9cVCgž`]4*mGS'W׽7^Cvn"8n6m1iv'1(s&i9U'Lqiߤ*٧dLaћ9x7@JKeBnFOשޚLyJ eډ4u~USDNNb"O2y0^(0y.8q$'6q:76XJ nmabd4Zo"ڙ2h%G4jC?^~{z>1?((ZC{eg;pD>TH}ڒdg lk3 SV9v?+6 s!ƖwⰉV*~r!њ +m [ D@RM&T֬' IG[Ō Px  c7$2M'C:ޒ$RNcߘuneۥl45/ ؄[ C+M-g$hr ˮGb[5VlQ|=#{Zӹr←5 xCÝnwǗ3kzQ˧vFR+GKϧ mၙ(rEXNڢ=0\Vј~ "w5; ra7.-'WO{jhd؝| n5{&8&9( 3JJc˽`޻Jt`xX  1fu9*1>%J,u;$C08EGFJN &, NTRcSYJřy$NmBjAX:~7˯(o@p=9A:N'Ģd DҼL@q|w]fai|fm.xErd1 AyfUwŁ,g`e{m*h&3 {bG o,Ys_I*%EpD/PK!H|Y&wagtail_graphql-0.1.2.dist-info/RECORDuɲH}= bFDAq $ ,<}߈갊<r(`pZ? /<m hv.)mSuF~?d^WF+4cCQ}հ_.ncy'Kf8YiۄV [;ZD_sJE\l}pEiqFgҊ)P19dܬՍSTte*ëǃφ*Cr&dtYg,f\iߊa,%(Zmx/hSy.cer!ݕ S b+ A_ĸP onJ/![_Â*c)gWeuz0HMUM^)@#Bqpwuz%nGWym|RUqٽ{wx,16\񤧷588fd vX:o$m Gkl LM)sTZ P-OEo,yAL=nIpw*{D0-0 ge|DkGrd& i'(rT}^Cf.8Kn\PQH^2tkw ؙ_Ⱥ@:l-\E{U~GA8٠$tdpdT:%68>W`ۅQ&S/z|8+T&OCYskI5z)4f[-s$A"+KμZQt؁29;˵YU#Gs=5|Gh6wݸ_<dz4%m6 |n˸Hz0E-APK!ewagtail_graphql/__init__.pyPK!jPwagtail_graphql/actions.pyPK!I,]];wagtail_graphql/apps.pyPK! wagtail_graphql/permissions.pyPK!ݗ,%wagtail_graphql/registry.pyPK!3AߩB-wagtail_graphql/schema.pyPK!"3wagtail_graphql/settings.pyPK!!5wagtail_graphql/types/__init__.pyPK!Q :wagtail_graphql/types/auth.pyPK!#<wagtail_graphql/types/core.pyPK!}PP"8Qwagtail_graphql/types/documents.pyPK!+99Vwagtail_graphql/types/forms.pyPK!\  =Xwagtail_graphql/types/images.pyPK!D~ewagtail_graphql/types/menus.pyPK!jj!jwagtail_graphql/types/settings.pyPK!8ش!nwagtail_graphql/types/snippets.pyPK!rvPy""$xqwagtail_graphql/types/streamfield.pyPK!X//'wagtail_graphql-0.1.2.dist-info/LICENSEPK!H|SU%.wagtail_graphql-0.1.2.dist-info/WHEELPK!H"Gk(Ęwagtail_graphql-0.1.2.dist-info/METADATAPK!H|Y&wagtail_graphql-0.1.2.dist-info/RECORDPKW