Aiohttp GraphQL server example

aiohttp_graphql_server_minimal_example.py

from aiohttp import web
from aiohttp_graphql import GraphQLView
from graphql.execution.executors.asyncio import AsyncioExecutor
from graphql import (
    graphql,
    GraphQLSchema,
    GraphQLObjectType,
    GraphQLField,
    GraphQLNonNull,
    GraphQLString,
    GraphQLInt,
    GraphQLList,
    GraphQLID
)


class Movie:

    def __init__(self, title, year):
        self.title = title
        self.year = year


class SampleModel:

    async def list_movies(self):
        return [
            Movie('The Matrix', 1999),
            Movie('The Matrix Reloaded', 2003),
            Movie('The Matrix Revolutions', 2003),
        ]


MovieType = GraphQLObjectType(
    name='Movie',
    fields={
        'title': GraphQLField(type=GraphQLString),
        'year': GraphQLField(type=GraphQLInt),
    }
)


async def resolve_movies(root, info):
    model = info.context['request'].app['model']
    return await model.list_movies()


Schema = GraphQLSchema(
    query=GraphQLObjectType(
        name='RootQueryType',
        fields={
            'movies': GraphQLField(
                type=GraphQLList(MovieType),
                resolver=resolve_movies),
        },
    ))


routes = web.RouteTableDef()


@routes.get('/')
async def index(request):
    return web.Response(
        text='Go to <a href="/graphql">/grphql</a>\n',
        content_type='text/html')


def get_app():
    app = web.Application()
    app.add_routes(routes)
    app['model'] = SampleModel()
    GraphQLView.attach(
        app,
        route_path='/graphql',
        schema=Schema,
        graphiql=True,
        executor=AsyncioExecutor())
    return app


app = get_app()


if __name__ == '__main__':
    web.run_app(app)