Moved django-graphql-view and django-graphiql into graphene-django 😎

This commit is contained in:
Syrus Akbary 2016-09-19 22:04:23 -07:00
parent a4306c890b
commit 3aa929fdc9
14 changed files with 1026 additions and 75 deletions

14
.gitignore vendored
View File

@ -59,20 +59,6 @@ docs/_build/
# PyBuilder
target/
/tests/django.sqlite
/graphene/index.json
/graphene/meta.json
/meta.json
/index.json
/docs/playground/graphene-js/pypyjs-release-nojit/
/docs/static/playground/lib
/docs/static/playground
# PyCharm
.idea

View File

@ -16,6 +16,33 @@ For instaling graphene, just run this command in your shell
pip install "graphene-django>=1.0.dev"
```
### Settings
```python
INSTALLED_APPS = (
# ...
'graphene_django',
)
GRAPHENE = {
'SCHEMA': 'app.schema.schema' # Where your Graphene schema lives
}
```
### Urls
We need to set up a `GraphQL` endpoint in our Django app, so we can serve the queries.
```python
from django.conf.urls import url
from graphene_django.views import GraphQLView
urlpatterns = [
# ...
url(r'^graphql', GraphQLView.as_view(graphiql=True)),
]
```
## Examples
Here is a simple Django model:

View File

@ -20,6 +20,36 @@ For instaling graphene, just run this command in your shell
pip install "graphene-django>=1.0.dev"
Settings
~~~~~~~~
.. code:: python
INSTALLED_APPS = (
# ...
'graphene_django',
)
GRAPHENE = {
'SCHEMA': 'app.schema.schema' # Where your Graphene schema lives
}
Urls
~~~~
We need to set up a ``GraphQL`` endpoint in our Django app, so we can
serve the queries.
.. code:: python
from django.conf.urls import url
from graphene_django.views import GraphQLView
urlpatterns = [
# ...
url(r'^graphql', GraphQLView.as_view(graphiql=True)),
]
Examples
--------

View File

@ -18,3 +18,17 @@ DATABASES = {
'NAME': 'django_test.sqlite',
}
}
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
'DIRS': [],
'APP_DIRS': True,
},
]
GRAPHENE = {
'SCHEMA': 'graphene_django.tests.schema_view.schema'
}
ROOT_URLCONF = 'graphene_django.tests.urls'

View File

@ -37,7 +37,7 @@ INSTALLED_APPS = [
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'django_graphiql',
'graphene_django',
'cookbook.ingredients.apps.IngredientsConfig',
'cookbook.recipes.apps.RecipesConfig',
@ -54,6 +54,10 @@ MIDDLEWARE_CLASSES = [
'django.middleware.clickjacking.XFrameOptionsMiddleware',
]
GRAPHENE = {
'SCHEMA': 'cookbook.schema.schema'
}
ROOT_URLCONF = 'cookbook.urls'
TEMPLATES = [

View File

@ -1,12 +1,10 @@
from django.conf.urls import include, url
from django.contrib import admin
from django.views.decorators.csrf import csrf_exempt
from cookbook.schema import schema
from graphene_django.views import GraphQLView
urlpatterns = [
url(r'^admin/', admin.site.urls),
url(r'^graphql', csrf_exempt(GraphQLView.as_view(schema=schema))),
url(r'^graphiql', include('django_graphiql.urls')),
url(r'^graphql', GraphQLView.as_view(graphiql=True)),
]

View File

@ -6,6 +6,8 @@ from optparse import make_option
from django import get_version as get_django_version
from django.core.management.base import BaseCommand, CommandError
from graphene_django.settings import graphene_settings
LT_DJANGO_1_8 = StrictVersion(get_django_version()) < StrictVersion('1.8')
if LT_DJANGO_1_8:
@ -16,7 +18,7 @@ if LT_DJANGO_1_8:
type=str,
dest='schema',
default='',
help='Django app containing schema to dump, e.g. myproject.core.schema',
help='Django app containing schema to dump, e.g. myproject.core.schema.schema',
),
make_option(
'--out',
@ -35,14 +37,14 @@ else:
'--schema',
type=str,
dest='schema',
default=getattr(settings, 'GRAPHENE_SCHEMA', ''),
help='Django app containing schema to dump, e.g. myproject.core.schema')
default=graphene_settings.SCHEMA,
help='Django app containing schema to dump, e.g. myproject.core.schema.schema')
parser.add_argument(
'--out',
type=str,
dest='out',
default=getattr(settings, 'GRAPHENE_SCHEMA_OUTPUT', 'schema.json'),
default=graphene_settings.SCHEMA_OUTPUT,
help='Output file (default: schema.json)')
@ -56,14 +58,18 @@ class Command(CommandArguments):
def handle(self, *args, **options):
from django.conf import settings
schema = options.get('schema') or getattr(settings, 'GRAPHENE_SCHEMA', '')
out = options.get('out') or getattr(settings, 'GRAPHENE_SCHEMA_OUTPUT', 'schema.json')
options_schema = options.get('schema')
if options_schema:
schema = importlib.import_module(options_schema)
else:
schema = graphene_settings.SCHEMA
if schema == '':
raise CommandError('Specify schema on GRAPHENE_SCHEMA setting or by using --schema')
i = importlib.import_module(schema)
out = options.get('out') or graphene_settings.SCHEMA_OUTPUT
schema_dict = {'data': i.schema.introspect()}
if not schema:
raise CommandError('Specify schema on GRAPHENE.SCHEMA setting or by using --schema')
schema_dict = {'data': schema.introspect()}
self.save_file(out, schema_dict)
style = getattr(self, 'style', None)

127
graphene_django/settings.py Normal file
View File

@ -0,0 +1,127 @@
"""
Settings for Graphene are all namespaced in the GRAPHENE setting.
For example your project's `settings.py` file might look like this:
GRAPHENE = {
'SCHEMA': 'my_app.schema.schema'
'MIDDLEWARE': (
'graphene_django.debug.DjangoDebugMiddleware',
)
}
This module provides the `graphene_settings` object, that is used to access
Graphene settings, checking for user settings first, then falling
back to the defaults.
"""
from __future__ import unicode_literals
from django.conf import settings
from django.test.signals import setting_changed
from django.utils import six
try:
import importlib # Available in Python 3.1+
except ImportError:
from django.utils import importlib # Will be removed in Django 1.9
# Copied shamelessly from Django REST Framework
DEFAULTS = {
'SCHEMA': None,
'SCHEMA_OUTPUT': 'schema.json',
'MIDDLEWARE': (),
}
if settings.DEBUG:
DEFAULTS['MIDDLEWARE'] += (
'graphene_django.debug.DjangoDebugMiddleware',
)
# List of settings that may be in string import notation.
IMPORT_STRINGS = (
'MIDDLEWARE',
'SCHEMA',
)
def perform_import(val, setting_name):
"""
If the given setting is a string import notation,
then perform the necessary import or imports.
"""
if val is None:
return None
elif isinstance(val, six.string_types):
return import_from_string(val, setting_name)
elif isinstance(val, (list, tuple)):
return [import_from_string(item, setting_name) for item in val]
return val
def import_from_string(val, setting_name):
"""
Attempt to import a class from a string representation.
"""
try:
# Nod to tastypie's use of importlib.
parts = val.split('.')
module_path, class_name = '.'.join(parts[:-1]), parts[-1]
module = importlib.import_module(module_path)
return getattr(module, class_name)
except (ImportError, AttributeError) as e:
msg = "Could not import '%s' for Graphene setting '%s'. %s: %s." % (val, setting_name, e.__class__.__name__, e)
raise ImportError(msg)
class GrapheneSettings(object):
"""
A settings object, that allows API settings to be accessed as properties.
For example:
from graphene_django.settings import settings
print(settings.SCHEMA)
Any setting with string import paths will be automatically resolved
and return the class, rather than the string literal.
"""
def __init__(self, user_settings=None, defaults=None, import_strings=None):
if user_settings:
self._user_settings = user_settings
self.defaults = defaults or DEFAULTS
self.import_strings = import_strings or IMPORT_STRINGS
@property
def user_settings(self):
if not hasattr(self, '_user_settings'):
self._user_settings = getattr(settings, 'GRAPHENE', {})
return self._user_settings
def __getattr__(self, attr):
if attr not in self.defaults:
raise AttributeError("Invalid Graphene setting: '%s'" % attr)
try:
# Check if present in user settings
val = self.user_settings[attr]
except KeyError:
# Fall back to defaults
val = self.defaults[attr]
# Coerce import strings into classes
if attr in self.import_strings:
val = perform_import(val, attr)
# Cache the result
setattr(self, attr, val)
return val
graphene_settings = GrapheneSettings(None, DEFAULTS, IMPORT_STRINGS)
def reload_graphene_settings(*args, **kwargs):
global graphene_settings
setting, value = kwargs['setting'], kwargs['value']
if setting == 'GRAPHENE':
graphene_settings = GrapheneSettings(value, DEFAULTS, IMPORT_STRINGS)
setting_changed.connect(reload_graphene_settings)

View File

@ -0,0 +1,123 @@
<!--
The request to this GraphQL server provided the header "Accept: text/html"
and as a result has been presented GraphiQL - an in-browser IDE for
exploring GraphQL.
If you wish to receive JSON, provide the header "Accept: application/json" or
add "&raw" to the end of the URL within a browser.
-->
<!DOCTYPE html>
<html>
<head>
<style>
html, body {
height: 100%;
margin: 0;
overflow: hidden;
width: 100%;
}
</style>
<link href="//cdn.jsdelivr.net/graphiql/{{graphiql_version}}/graphiql.css" rel="stylesheet" />
<script src="//cdn.jsdelivr.net/fetch/0.9.0/fetch.min.js"></script>
<script src="//cdn.jsdelivr.net/react/15.0.1/react.min.js"></script>
<script src="//cdn.jsdelivr.net/react/15.0.1/react-dom.min.js"></script>
<script src="//cdn.jsdelivr.net/graphiql/{{graphiql_version}}/graphiql.min.js"></script>
</head>
<body>
<script>
// Parse the cookie value for a CSRF token
var csrftoken;
var cookies = ('; ' + document.cookie).split('; csrftoken=');
if (cookies.length == 2)
csrftoken = cookies.pop().split(';').shift();
// Collect the URL parameters
var parameters = {};
window.location.search.substr(1).split('&').forEach(function (entry) {
var eq = entry.indexOf('=');
if (eq >= 0) {
parameters[decodeURIComponent(entry.slice(0, eq))] =
decodeURIComponent(entry.slice(eq + 1));
}
});
// Produce a Location query string from a parameter object.
function locationQuery(params) {
return '?' + Object.keys(params).map(function (key) {
return encodeURIComponent(key) + '=' +
encodeURIComponent(params[key]);
}).join('&');
}
// Derive a fetch URL from the current URL, sans the GraphQL parameters.
var graphqlParamNames = {
query: true,
variables: true,
operationName: true
};
var otherParams = {};
for (var k in parameters) {
if (parameters.hasOwnProperty(k) && graphqlParamNames[k] !== true) {
otherParams[k] = parameters[k];
}
}
var fetchURL = locationQuery(otherParams);
// Defines a GraphQL fetcher using the fetch API.
function graphQLFetcher(graphQLParams) {
var headers = {
'Accept': 'application/json',
'Content-Type': 'application/json'
};
if (csrftoken) {
headers['X-CSRFToken'] = csrftoken;
}
return fetch(fetchURL, {
method: 'post',
headers: headers,
body: JSON.stringify(graphQLParams),
credentials: 'include',
}).then(function (response) {
return response.text();
}).then(function (responseBody) {
try {
return JSON.parse(responseBody);
} catch (error) {
return responseBody;
}
});
}
// When the query and variables string is edited, update the URL bar so
// that it can be easily shared.
function onEditQuery(newQuery) {
parameters.query = newQuery;
updateURL();
}
function onEditVariables(newVariables) {
parameters.variables = newVariables;
updateURL();
}
function onEditOperationName(newOperationName) {
parameters.operationName = newOperationName;
updateURL();
}
function updateURL() {
history.replaceState(null, null, locationQuery(parameters));
}
// Render <GraphiQL /> into the body.
ReactDOM.render(
React.createElement(GraphiQL, {
fetcher: graphQLFetcher,
onEditQuery: onEditQuery,
onEditVariables: onEditVariables,
onEditOperationName: onEditOperationName,
query: '{{ query|escapejs }}',
response: '{{ result|escapejs }}',
{% if variables %}
variables: '{{ variables|escapejs }}',
{% endif %}
{% if operationName %}
operationName: '{{ operation_name|escapejs }}',
{% endif %}
}),
document.body
);
</script>
</body>
</html>

View File

@ -0,0 +1,32 @@
import graphene
from graphene import Schema, ObjectType, relay
from ..types import DjangoObjectType
from .models import Article, Reporter
class QueryRoot(ObjectType):
thrower = graphene.String(required=True)
request = graphene.String(required=True)
test = graphene.String(who=graphene.String())
def resolve_thrower(self, args, context, info):
raise Exception("Throws!")
def resolve_request(self, args, context, info):
request = context
return request.GET.get('q')
def resolve_test(self, args, context, info):
return 'Hello %s' % (args.get('who') or 'World')
class MutationRoot(ObjectType):
write_test = graphene.Field(QueryRoot)
def resolve_write_test(self, args, context, info):
return QueryRoot()
schema = Schema(query=QueryRoot, mutation=MutationRoot)

View File

@ -5,7 +5,6 @@ from six import StringIO
@patch('graphene_django.management.commands.graphql_schema.Command.save_file')
def test_generate_file_on_call_graphql_schema(savefile_mock, settings):
settings.GRAPHENE_SCHEMA = 'graphene_django.tests.schema'
out = StringIO()
management.call_command('graphql_schema', schema='', stdout=out)
assert "Successfully dumped GraphQL schema to schema.json" in out.getvalue()

View File

@ -1,57 +1,421 @@
import pytest
import json
try:
from urllib import urlencode
except ImportError:
from urllib.parse import urlencode
def format_response(response):
def url_string(**url_params):
string = '/graphql'
if url_params:
string += '?' + urlencode(url_params)
return string
def response_json(response):
return json.loads(response.content.decode())
def test_client_get_good_query(settings, client):
settings.ROOT_URLCONF = 'graphene_django.tests.urls'
response = client.get('/graphql', {'query': '{ human { headline } }'})
json_response = format_response(response)
expected_json = {
j = lambda **kwargs: json.dumps(kwargs)
def test_graphiql_is_enabled(client):
response = client.get(url_string(), HTTP_ACCEPT='text/html')
assert response.status_code == 200
def test_allows_get_with_query_param(client):
response = client.get(url_string(query='{test}'))
assert response.status_code == 200
assert response_json(response) == {
'data': {'test': "Hello World"}
}
def test_allows_get_with_variable_values(client):
response = client.get(url_string(
query='query helloWho($who: String){ test(who: $who) }',
variables=json.dumps({'who': "Dolly"})
))
assert response.status_code == 200
assert response_json(response) == {
'data': {'test': "Hello Dolly"}
}
def test_allows_get_with_operation_name(client):
response = client.get(url_string(
query='''
query helloYou { test(who: "You"), ...shared }
query helloWorld { test(who: "World"), ...shared }
query helloDolly { test(who: "Dolly"), ...shared }
fragment shared on QueryRoot {
shared: test(who: "Everyone")
}
''',
operationName='helloWorld'
))
assert response.status_code == 200
assert response_json(response) == {
'data': {
'human': {
'headline': None
}
'test': 'Hello World',
'shared': 'Hello Everyone'
}
}
assert json_response == expected_json
def test_client_get_good_query_with_raise(settings, client):
settings.ROOT_URLCONF = 'graphene_django.tests.urls'
response = client.get('/graphql', {'query': '{ human { raises } }'})
json_response = format_response(response)
assert json_response['errors'][0]['message'] == 'This field should raise exception'
assert json_response['data']['human']['raises'] is None
def test_reports_validation_errors(client):
response = client.get(url_string(
query='{ test, unknownOne, unknownTwo }'
))
def test_client_post_good_query_json(settings, client):
settings.ROOT_URLCONF = 'graphene_django.tests.urls'
response = client.post(
'/graphql', json.dumps({'query': '{ human { headline } }'}), 'application/json')
json_response = format_response(response)
expected_json = {
'data': {
'human': {
'headline': None
assert response.status_code == 400
assert response_json(response) == {
'errors': [
{
'message': 'Cannot query field "unknownOne" on type "QueryRoot".',
'locations': [{'line': 1, 'column': 9}]
},
{
'message': 'Cannot query field "unknownTwo" on type "QueryRoot".',
'locations': [{'line': 1, 'column': 21}]
}
]
}
def test_errors_when_missing_operation_name(client):
response = client.get(url_string(
query='''
query TestQuery { test }
mutation TestMutation { writeTest { test } }
'''
))
assert response.status_code == 400
assert response_json(response) == {
'errors': [
{
'message': 'Must provide operation name if query contains multiple operations.'
}
]
}
def test_errors_when_sending_a_mutation_via_get(client):
response = client.get(url_string(
query='''
mutation TestMutation { writeTest { test } }
'''
))
assert response.status_code == 405
assert response_json(response) == {
'errors': [
{
'message': 'Can only perform a mutation operation from a POST request.'
}
]
}
def test_errors_when_selecting_a_mutation_within_a_get(client):
response = client.get(url_string(
query='''
query TestQuery { test }
mutation TestMutation { writeTest { test } }
''',
operationName='TestMutation'
))
assert response.status_code == 405
assert response_json(response) == {
'errors': [
{
'message': 'Can only perform a mutation operation from a POST request.'
}
]
}
def test_allows_mutation_to_exist_within_a_get(client):
response = client.get(url_string(
query='''
query TestQuery { test }
mutation TestMutation { writeTest { test } }
''',
operationName='TestQuery'
))
assert response.status_code == 200
assert response_json(response) == {
'data': {'test': "Hello World"}
}
def test_allows_post_with_json_encoding(client):
response = client.post(url_string(), j(query='{test}'), 'application/json')
assert response.status_code == 200
assert response_json(response) == {
'data': {'test': "Hello World"}
}
def test_allows_sending_a_mutation_via_post(client):
response = client.post(url_string(), j(query='mutation TestMutation { writeTest { test } }'), 'application/json')
assert response.status_code == 200
assert response_json(response) == {
'data': {'writeTest': {'test': 'Hello World'}}
}
def test_allows_post_with_url_encoding(client):
response = client.post(url_string(), urlencode(dict(query='{test}')), 'application/x-www-form-urlencoded')
assert response.status_code == 200
assert response_json(response) == {
'data': {'test': "Hello World"}
}
def test_supports_post_json_query_with_string_variables(client):
response = client.post(url_string(), j(
query='query helloWho($who: String){ test(who: $who) }',
variables=json.dumps({'who': "Dolly"})
), 'application/json')
assert response.status_code == 200
assert response_json(response) == {
'data': {'test': "Hello Dolly"}
}
def test_supports_post_json_query_with_json_variables(client):
response = client.post(url_string(), j(
query='query helloWho($who: String){ test(who: $who) }',
variables={'who': "Dolly"}
), 'application/json')
assert response.status_code == 200
assert response_json(response) == {
'data': {'test': "Hello Dolly"}
}
def test_supports_post_url_encoded_query_with_string_variables(client):
response = client.post(url_string(), urlencode(dict(
query='query helloWho($who: String){ test(who: $who) }',
variables=json.dumps({'who': "Dolly"})
)), 'application/x-www-form-urlencoded')
assert response.status_code == 200
assert response_json(response) == {
'data': {'test': "Hello Dolly"}
}
def test_supports_post_json_quey_with_get_variable_values(client):
response = client.post(url_string(
variables=json.dumps({'who': "Dolly"})
), j(
query='query helloWho($who: String){ test(who: $who) }',
), 'application/json')
assert response.status_code == 200
assert response_json(response) == {
'data': {'test': "Hello Dolly"}
}
def test_post_url_encoded_query_with_get_variable_values(client):
response = client.post(url_string(
variables=json.dumps({'who': "Dolly"})
), urlencode(dict(
query='query helloWho($who: String){ test(who: $who) }',
)), 'application/x-www-form-urlencoded')
assert response.status_code == 200
assert response_json(response) == {
'data': {'test': "Hello Dolly"}
}
def test_supports_post_raw_text_query_with_get_variable_values(client):
response = client.post(url_string(
variables=json.dumps({'who': "Dolly"})
),
'query helloWho($who: String){ test(who: $who) }',
'application/graphql'
)
assert response.status_code == 200
assert response_json(response) == {
'data': {'test': "Hello Dolly"}
}
def test_allows_post_with_operation_name(client):
response = client.post(url_string(), j(
query='''
query helloYou { test(who: "You"), ...shared }
query helloWorld { test(who: "World"), ...shared }
query helloDolly { test(who: "Dolly"), ...shared }
fragment shared on QueryRoot {
shared: test(who: "Everyone")
}
''',
operationName='helloWorld'
), 'application/json')
assert response.status_code == 200
assert response_json(response) == {
'data': {
'test': 'Hello World',
'shared': 'Hello Everyone'
}
}
assert json_response == expected_json
def test_client_post_good_query_graphql(settings, client):
settings.ROOT_URLCONF = 'graphene_django.tests.urls'
response = client.post(
'/graphql', '{ human { headline } }', 'application/graphql')
json_response = format_response(response)
expected_json = {
def test_allows_post_with_get_operation_name(client):
response = client.post(url_string(
operationName='helloWorld'
), '''
query helloYou { test(who: "You"), ...shared }
query helloWorld { test(who: "World"), ...shared }
query helloDolly { test(who: "Dolly"), ...shared }
fragment shared on QueryRoot {
shared: test(who: "Everyone")
}
''',
'application/graphql')
assert response.status_code == 200
assert response_json(response) == {
'data': {
'human': {
'headline': None
}
'test': 'Hello World',
'shared': 'Hello Everyone'
}
}
@pytest.mark.urls('tests.urls_pretty')
def test_supports_pretty_printing(client):
response = client.get(url_string(query='{test}'))
assert response.content.decode() == (
'{\n'
' "data": {\n'
' "test": "Hello World"\n'
' }\n'
'}'
)
def test_supports_pretty_printing_by_request(client):
response = client.get(url_string(query='{test}', pretty='1'))
assert response.content.decode() == (
'{\n'
' "data": {\n'
' "test": "Hello World"\n'
' }\n'
'}'
)
def test_handles_field_errors_caught_by_graphql(client):
response = client.get(url_string(query='{thrower}'))
assert response.status_code == 200
assert response_json(response) == {
'data': None,
'errors': [{'locations': [{'column': 2, 'line': 1}], 'message': 'Throws!'}]
}
def test_handles_syntax_errors_caught_by_graphql(client):
response = client.get(url_string(query='syntaxerror'))
assert response.status_code == 400
assert response_json(response) == {
'errors': [{'locations': [{'column': 1, 'line': 1}],
'message': 'Syntax Error GraphQL request (1:1) '
'Unexpected Name "syntaxerror"\n\n1: syntaxerror\n ^\n'}]
}
def test_handles_errors_caused_by_a_lack_of_query(client):
response = client.get(url_string())
assert response.status_code == 400
assert response_json(response) == {
'errors': [{'message': 'Must provide query string.'}]
}
def test_handles_invalid_json_bodies(client):
response = client.post(url_string(), '[]', 'application/json')
assert response.status_code == 400
assert response_json(response) == {
'errors': [{'message': 'POST body sent invalid JSON.'}]
}
def test_handles_incomplete_json_bodies(client):
response = client.post(url_string(), '{"query":', 'application/json')
assert response.status_code == 400
assert response_json(response) == {
'errors': [{'message': 'POST body sent invalid JSON.'}]
}
def test_handles_plain_post_text(client):
response = client.post(url_string(
variables=json.dumps({'who': "Dolly"})
),
'query helloWho($who: String){ test(who: $who) }',
'text/plain'
)
assert response.status_code == 400
assert response_json(response) == {
'errors': [{'message': 'Must provide query string.'}]
}
def test_handles_poorly_formed_variables(client):
response = client.get(url_string(
query='query helloWho($who: String){ test(who: $who) }',
variables='who:You'
))
assert response.status_code == 400
assert response_json(response) == {
'errors': [{'message': 'Variables are invalid JSON.'}]
}
def test_handles_unsupported_http_methods(client):
response = client.put(url_string(query='{test}'))
assert response.status_code == 405
assert response['Allow'] == 'GET, POST'
assert response_json(response) == {
'errors': [{'message': 'GraphQL only supports GET and POST requests.'}]
}
def test_passes_request_into_context_request(client):
response = client.get(url_string(query='{request}', q='testing'))
assert response.status_code == 200
assert response_json(response) == {
'data': {
'request': 'testing'
}
}
assert json_response == expected_json

View File

@ -1,8 +1,7 @@
from django.conf.urls import url
from ..views import GraphQLView
from .schema import schema
urlpatterns = [
url(r'^graphql', GraphQLView.as_view(schema=schema)),
url(r'^graphql', GraphQLView.as_view(graphiql=True)),
]

View File

@ -1,10 +1,252 @@
from graphql_django_view import GraphQLView as BaseGraphQLView
import re
import json
import six
from django.http import HttpResponse, HttpResponseNotAllowed
from django.http.response import HttpResponseBadRequest
from django.views.generic import View
from django.shortcuts import render
from graphql import Source, parse, execute, validate
from graphql.error import GraphQLError, format_error as format_graphql_error
from graphql.execution import ExecutionResult
from graphql.type.schema import GraphQLSchema
from graphql.execution.middleware import MiddlewareManager
from graphql.utils.get_operation_ast import get_operation_ast
from .settings import graphene_settings
class GraphQLView(BaseGraphQLView):
class HttpError(Exception):
def __init__(self, response, message=None, *args, **kwargs):
self.response = response
self.message = message = message or response.content.decode()
super(HttpError, self).__init__(message, *args, **kwargs)
def __init__(self, schema, **kwargs):
super(GraphQLView, self).__init__(
schema=schema,
**kwargs
)
def get_accepted_content_types(request):
def qualify(x):
parts = x.split(';', 1)
if len(parts) == 2:
match = re.match(r'(^|;)q=(0(\.\d{,3})?|1(\.0{,3})?)(;|$)',
parts[1])
if match:
return parts[0], float(match.group(2))
return parts[0], 1
raw_content_types = request.META.get('HTTP_ACCEPT', '*/*').split(',')
qualified_content_types = map(qualify, raw_content_types)
return list(x[0] for x in sorted(qualified_content_types,
key=lambda x: x[1], reverse=True))
class GraphQLView(View):
graphiql_version = '0.7.8'
graphiql_template = 'graphene/graphiql.html'
graphiql = False
executor = None
middleware = None
root_value = None
pretty = False
def __init__(self, schema=None, executor=None, middleware=None, root_value=None, graphiql=False, pretty=False):
if not schema:
schema = graphene_settings.SCHEMA
if middleware is None:
middleware = graphene_settings.MIDDLEWARE
self.schema = schema
self.middleware = MiddlewareManager(middleware)
self.executor = executor
self.root_value = root_value
self.pretty = pretty
self.graphiql = graphiql
assert isinstance(self.schema, GraphQLSchema), 'A Schema is required to be provided to GraphQLView.'
# noinspection PyUnusedLocal
def get_root_value(self, request):
return self.root_value
def get_middleware(self, request):
return self.middleware
def get_context(self, request):
return request
def dispatch(self, request, *args, **kwargs):
try:
if request.method.lower() not in ('get', 'post'):
raise HttpError(HttpResponseNotAllowed(['GET', 'POST'], 'GraphQL only supports GET and POST requests.'))
data = self.parse_body(request)
show_graphiql = self.graphiql and self.can_display_graphiql(request, data)
query, variables, operation_name = self.get_graphql_params(request, data)
execution_result = self.execute_graphql_request(
request,
data,
query,
variables,
operation_name,
show_graphiql
)
if execution_result:
response = {}
if execution_result.errors:
response['errors'] = [self.format_error(e) for e in execution_result.errors]
if execution_result.invalid:
status_code = 400
else:
status_code = 200
response['data'] = execution_result.data
result = self.json_encode(request, response, pretty=show_graphiql)
else:
result = None
if show_graphiql:
return self.render_graphiql(
request,
graphiql_version=self.graphiql_version,
query=query or '',
variables=variables or '',
operation_name=operation_name or '',
result=result or ''
)
return HttpResponse(
status=status_code,
content=result,
content_type='application/json'
)
except HttpError as e:
response = e.response
response['Content-Type'] = 'application/json'
response.content = self.json_encode(request, {
'errors': [self.format_error(e)]
})
return response
def render_graphiql(self, request, **data):
return render(request, self.graphiql_template, data)
def json_encode(self, request, d, pretty=False):
if not (self.pretty or pretty) and not request.GET.get('pretty'):
return json.dumps(d, separators=(',', ':'))
return json.dumps(d, sort_keys=True,
indent=2, separators=(',', ': '))
# noinspection PyBroadException
def parse_body(self, request):
content_type = self.get_content_type(request)
if content_type == 'application/graphql':
return {'query': request.body.decode()}
elif content_type == 'application/json':
try:
request_json = json.loads(request.body.decode('utf-8'))
assert isinstance(request_json, dict)
return request_json
except:
raise HttpError(HttpResponseBadRequest('POST body sent invalid JSON.'))
elif content_type in ['application/x-www-form-urlencoded', 'multipart/form-data']:
return request.POST
return {}
def execute(self, *args, **kwargs):
return execute(self.schema, *args, **kwargs)
def execute_graphql_request(self, request, data, query, variables, operation_name, show_graphiql=False):
if not query:
if show_graphiql:
return None
raise HttpError(HttpResponseBadRequest('Must provide query string.'))
source = Source(query, name='GraphQL request')
try:
document_ast = parse(source)
validation_errors = validate(self.schema, document_ast)
if validation_errors:
return ExecutionResult(
errors=validation_errors,
invalid=True,
)
except Exception as e:
return ExecutionResult(errors=[e], invalid=True)
if request.method.lower() == 'get':
operation_ast = get_operation_ast(document_ast, operation_name)
if operation_ast and operation_ast.operation != 'query':
if show_graphiql:
return None
raise HttpError(HttpResponseNotAllowed(
['POST'], 'Can only perform a {} operation from a POST request.'.format(operation_ast.operation)
))
try:
return self.execute(
document_ast,
root_value=self.get_root_value(request),
variable_values=variables,
operation_name=operation_name,
context_value=self.get_context(request),
middlewares=self.get_middleware(request),
executor=self.executor,
)
except Exception as e:
return ExecutionResult(errors=[e], invalid=True)
@classmethod
def can_display_graphiql(cls, request, data):
raw = 'raw' in request.GET or 'raw' in data
return not raw and cls.request_wants_html(request)
@classmethod
def request_wants_html(cls, request):
accepted = get_accepted_content_types(request)
html_index = accepted.count('text/html')
json_index = accepted.count('application/json')
return html_index > json_index
@staticmethod
def get_graphql_params(request, data):
query = request.GET.get('query') or data.get('query')
variables = request.GET.get('variables') or data.get('variables')
if variables and isinstance(variables, six.text_type):
try:
variables = json.loads(variables)
except:
raise HttpError(HttpResponseBadRequest('Variables are invalid JSON.'))
operation_name = request.GET.get('operationName') or data.get('operationName')
return query, variables, operation_name
@staticmethod
def format_error(error):
if isinstance(error, GraphQLError):
return format_graphql_error(error)
return {'message': six.text_type(error)}
@staticmethod
def get_content_type(request):
meta = request.META
content_type = meta.get('CONTENT_TYPE', meta.get('HTTP_CONTENT_TYPE', ''))
return content_type.split(';', 1)[0].lower()