converter.py

from django import forms
from django.forms import fields
from singledispatch import singledispatch

from graphene.core.types.scalars import ID, Boolean, Float, Int, String
from graphene.contrib.django.fields import ConnectionOrListField, DjangoModelField

try:
    UUIDField = forms.UUIDField
except AttributeError:
    # Improved compatibility for Django 1.6
    class UUIDField(object):
        pass

@singledispatch
def convert_form_field(field):
    raise Exception(
        "Don't know how to convert the Django form field %s (%s) "
        "to Graphene type" %
        (field, field.__class__)
    )


@convert_form_field.register(fields.BaseTemporalField)
@convert_form_field.register(fields.CharField)
@convert_form_field.register(fields.EmailField)
@convert_form_field.register(fields.SlugField)
@convert_form_field.register(fields.URLField)
@convert_form_field.register(UUIDField)
def convert_form_field_to_string(field):
    return String(description=field.help_text)


@convert_form_field.register(fields.IntegerField)
@convert_form_field.register(fields.NumberInput)
def convert_form_field_to_int(field):
    return Int(description=field.help_text)


@convert_form_field.register(fields.BooleanField)
@convert_form_field.register(fields.NullBooleanField)
def convert_form_field_to_boolean(field):
    return Boolean(description=field.help_text, required=True)


@convert_form_field.register(fields.NullBooleanField)
def convert_form_field_to_nullboolean(field):
    return Boolean(description=field.help_text)


@convert_form_field.register(fields.DecimalField)
@convert_form_field.register(fields.FloatField)
def convert_form_field_to_float(field):
    return Float(description=field.help_text)


@convert_form_field.register(forms.ModelMultipleChoiceField)
def convert_form_field_to_list_or_connection(field):
    model_field = DjangoModelField(field.related_model)
    return ConnectionOrListField(model_field)


@convert_form_field.register(forms.ModelChoiceField)
def convert_form_field_to_djangomodel(field):
    #TODO: Probably needs some work
    return ID()
    # return DjangoModelField(field.queryset.model, description=field.help_text)

filters.py

import django_filters

from .models import Account, Transaction


class AccountFilter(django_filters.FilterSet):
    class Meta:
        model = Account
        fields = {
            'name': ['exact', 'icontains'],
            'slug': ['exact'],
            'created': ['gt', 'lt', 'exact'],
            # 'owned_by__username': ['exact'],
        }
        order_by = True


class TransactionFilter(django_filters.FilterSet):
    has_bill = django_filters.BooleanFilter(name='bill', lookup_type='isnull', exclude=True)

    class Meta:
        model = Transaction
        fields = {
            'amount': ['gt', 'lt', 'exact'],
            # TODO: Won't work until we have implemented a GlobalIDFilter + GlobalIDField
            'bill': ['exact'],
        }
        order_by = True

nodes.py

from django.contrib.auth import get_user_model
import graphene
from graphene.contrib.django import DjangoNode
from graphene.contrib.django.fields import DjangoConnectionField, DjangoModelField
from swiftwind.accounts import models
from swiftwind.bills.models import Bill
from swiftwind.schema_fields import DjangoFilterConnectionField
from swiftwind.schema_resolvers import SimpleQuerySetConnectionResolver

from ..filters import TransactionFilter

class Transaction(DjangoNode):

    class Meta:
        model = models.Transaction


class Account(DjangoNode):
    balance = graphene.Int()
    transactions_in = DjangoFilterConnectionField(Transaction, on='transactions_in', filterset_class=TransactionFilter)
    transactions_out = DjangoFilterConnectionField(Transaction, on='transactions_out', filterset_class=TransactionFilter)

    class Meta:
        model = models.Account

    def resolve_balance(self, args, info):
        return self.calculate_balance()


class User(DjangoNode):
    full_name = graphene.StringField()

    class Meta:
        model = get_user_model()
        only_fields = ('username', 'first_name', 'last_name', 'email', 'is_active')

    def resolve_full_name(self, args, info):
        return self.get_full_name()

schema.py

import graphene
from graphene import relay
from graphene import ObjectType

import swiftwind.accounts.schema.nodes as accounts_nodes
import swiftwind.bills.schema.nodes as bills_nodes
from swiftwind.accounts.filters import AccountFilter, TransactionFilter
from swiftwind.bills.filters import BillFilter
from swiftwind.schema_resolvers import SimpleQuerySetConnectionResolver
from swiftwind.schema_fields import DjangoFilterConnectionField

schema = graphene.Schema(name='Swiftwind Schema')


class Query(ObjectType):
    account = relay.NodeField(accounts_nodes.Account)
    all_accounts = DjangoFilterConnectionField(accounts_nodes.Account,
                                               filterset_class=AccountFilter)

    transaction = relay.NodeField(accounts_nodes.Transaction)
    all_transactions = DjangoFilterConnectionField(accounts_nodes.Transaction,
                                                   filterset_class=TransactionFilter)

    user = relay.NodeField(accounts_nodes.User)
    all_users = relay.ConnectionField(accounts_nodes.User,
                                      resolver=SimpleQuerySetConnectionResolver(accounts_nodes.User))

schema.query = Query

schema_fields.py

import graphene
import six
from graphene.contrib.django import DjangoConnectionField

from swiftwind.converter import convert_form_field
from swiftwind.schema_resolvers import FilterConnectionResolver


class DjangoFilterConnectionField(DjangoConnectionField):

    def __init__(self, type, filterset_class, resolver=None, on=None, *args, **kwargs):
        if not resolver:
            resolver = FilterConnectionResolver(type, on, filterset_class)

        kwargs.setdefault('args', {})
        kwargs['args'].update(**self.get_filtering_args(type, filterset_class))
        super(DjangoFilterConnectionField, self).__init__(type, resolver, *args, **kwargs)

    def get_filtering_args(self, type, filterset_class):
        args = {}
        for name, filter_field in six.iteritems(filterset_class.base_filters):
            field_type = graphene.Argument(convert_form_field(filter_field.field))
            # Is this correct? I don't quite grok the 'parent' system yet
            field_type.mount(type)
            args[name] = field_type

        # Also add the 'order_by' field
        args[filterset_class.order_by_field] = graphene.Argument(graphene.String)
        return args

schema_resolvers.py

from django.core.exceptions import ImproperlyConfigured
from django_filters.filterset import filterset_factory


class BaseQuerySetConnectionResolver(object):

    def __init__(self, node, on=None):
        self.node = node
        self.model = node._meta.model
        # The name of the field on the model which contains the
        # manager upon which to perform the query. Optional.
        # If omitted the model's default manager will be used.
        self.on = on

    def __call__(self, inst, args, info):
        self.inst = inst
        self.args = args
        self.info = info
        return self.make_query()

    def get_manager(self):
        if self.on:
            return getattr(self.inst, self.on)
        else:
            return self.model._default_manager

    def make_query(self):
        raise NotImplemented()


class SimpleQuerySetConnectionResolver(BaseQuerySetConnectionResolver):
    # Simple querying without using django-filter (ported from previous gist)

    def make_query(self):
        filter_kwargs = self.get_filter_kwargs()
        query = self.get_manager().filter(**filter_kwargs)
        order = self.get_order()
        if order:
            query = query.order_by(order)
        return query

    def get_filter_kwargs(self):
        ignore = ['first', 'last', 'before', 'after', 'order']
        return {k: v for k, v in self.args.items() if k not in ignore}

    def get_order(self):
        return self.args.get('order', None)


class FilterConnectionResolver(BaseQuerySetConnectionResolver):
    # Querying using django-filter

    def __init__(self, node, on=None, filterset_class=None):
        self.filterset_class = filterset_class
        super(FilterConnectionResolver, self).__init__(node, on)

    def make_query(self):
        filterset_class = self.get_filterset_class()
        filterset = self.get_filterset(filterset_class)
        return filterset.qs

    def get_filterset_class(self):
        if self.filterset_class:
            return self.filterset_class
        elif self.model:
            return filterset_factory(self.model)
        else:
            msg = "'%s' must define 'filterset_class' or 'model'"
            raise ImproperlyConfigured(msg % self.__class__.__name__)

    def get_filterset(self, filterset_class):
        kwargs = self.get_filterset_kwargs(filterset_class)
        return filterset_class(**kwargs)

    def get_filterset_kwargs(self, filterset_class):
        kwargs = {'data': self.args or None}
        try:
            kwargs.update({
                'queryset': self.get_manager(),
            })
        except ImproperlyConfigured:
            # ignore the error here if the filterset has a model defined
            # to acquire a queryset from
            if filterset_class._meta.model is None:
                msg = ("'%s' does not define a 'model' and the resolver '%s' "
                       "does not return a valid queryset from 'get_queryset'. "
                       "You must fix one of them.")
                args = (filterset_class.__name__, self.__class__.__name__)
                raise ImproperlyConfigured(msg % args)
        return kwargs