Graphene - Graphql on Python

Introduction to GraphQL and Graphene

GraphQL — is a modern query language for APIs and a runtime, developed by Facebook in 2012 to solve the problems of traditional REST APIs. Graphene — is a powerful Python library that provides a full implementation of the GraphQL specification with support for typed schemas, mutations, and integration with popular ORM systems.

What is GraphQL and why you need it

GraphQL is a declarative approach to data fetching, where the client precisely describes the data it needs. Unlike REST, where the server defines the response structure, GraphQL allows the client to shape the request and receive exactly the data required.

Key advantages of GraphQL over REST

Single entry point

GraphQL uses a single URL endpoint for all operations, simplifying API architecture and eliminating the need for versioning via URLs.

Query flexibility

The client defines the shape of the response, requesting only the necessary fields and related data.

Elimination of data redundancy

Over-fetching (getting unnecessary data) and under-fetching (getting insufficient data) are resolved at the query level.

Combining related data

Ability to retrieve data from multiple related entities in a single request, reducing the number of server round‑trips.

Strong typing

The GraphQL schema is statically typed, providing request validation and IDE auto‑completion.

Automatic documentation

The schema serves as live API documentation, accessible via introspection.

Installation and setup of Graphene

Basic installation

pip install graphene

Integration with web frameworks

For Flask:

pip install flask-graphql

For Django:

pip install graphene-django

For FastAPI:

pip install starlette-graphene3

Database integration

For SQLAlchemy:

pip install graphene-sqlalchemy

For Django ORM:

pip install graphene-django

Creating a basic GraphQL server

Flask example

from flask import Flask
from flask_graphql import GraphQLView
import graphene

app = Flask(__name__)

class Query(graphene.ObjectType):
    hello = graphene.String(name=graphene.String(default_value="world"))
    
    def resolve_hello(self, info, name):
        return f"Hello, {name}!"

schema = graphene.Schema(query=Query)

app.add_url_rule('/graphql', view_func=GraphQLView.as_view(
    'graphql', 
    schema=schema, 
    graphiql=True  # Enables the GraphiQL interface
))

if __name__ == '__main__':
    app.run(debug=True)

Django example

# settings.py
INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'graphene_django',
    'myapp',
]

GRAPHENE = {
    'SCHEMA': 'myproject.schema.schema'
}

# urls.py
from django.urls import path
from graphene_django.views import GraphQLView

urlpatterns = [
    path('graphql/', GraphQLView.as_view(graphiql=True)),
]

Defining types and schemas in Graphene

Creating custom types

import graphene

class User(graphene.ObjectType):
    id = graphene.ID()
    name = graphene.String()
    email = graphene.String()
    age = graphene.Int()
    is_active = graphene.Boolean()
    
    # Computed fields
    full_name = graphene.String()
    
    def resolve_full_name(self, info):
        return f"{self.first_name} {self.last_name}"

class Post(graphene.ObjectType):
    id = graphene.ID()
    title = graphene.String()
    content = graphene.String()
    author = graphene.Field(User)
    created_at = graphene.DateTime()

Defining the query schema

class Query(graphene.ObjectType):
    user = graphene.Field(User, id=graphene.ID(required=True))
    users = graphene.List(User, limit=graphene.Int(default_value=10))
    
    def resolve_user(self, info, id):
        # Logic for fetching a user by ID
        return User(id=id, name="Ivan Petrov", email="ivan@example.com")
    
    def resolve_users(self, info, limit):
        # Logic for fetching a list of users
        return [User(id=i, name=f"User {i}") for i in range(1, limit+1)]

Queries: structure and handling

Simple queries

{
  user(id: "1") {
    id
    name
    email
  }
}

Queries with aliases

{
  firstUser: user(id: "1") {
    name
  }
  secondUser: user(id: "2") {
    name
  }
}

Nested queries

class User(graphene.ObjectType):
    id = graphene.ID()
    name = graphene.String()
    posts = graphene.List(lambda: Post)
    
    def resolve_posts(self, info):
        # Fetch user posts
        return [Post(title="First post"), Post(title="Second post")]

{
  user(id: "1") {
    name
    posts {
      title
      content
    }
  }
}

Mutations: modifying data

Creating a mutation

class CreateUser(graphene.Mutation):
    class Arguments:
        name = graphene.String(required=True)
        email = graphene.String(required=True)
    
    # Return fields
    ok = graphene.Boolean()
    user = graphene.Field(User)
    errors = graphene.List(graphene.String)
    
    def mutate(self, info, name, email):
        # Validation
        if not email or '@' not in email:
            return CreateUser(ok=False, errors=["Invalid email"])
        
        # Create user
        user = User(name=name, email=email)
        return CreateUser(ok=True, user=user)

class UpdateUser(graphene.Mutation):
    class Arguments:
        id = graphene.ID(required=True)
        name = graphene.String()
        email = graphene.String()
    
    user = graphene.Field(User)
    
    def mutate(self, info, id, name=None, email=None):
        # Update logic
        user = User(id=id, name=name, email=email)
        return UpdateUser(user=user)

class Mutation(graphene.ObjectType):
    create_user = CreateUser.Field()
    update_user = UpdateUser.Field()

Executing a mutation

mutation {
  createUser(name: "Ivan Petrov", email: "ivan@example.com") {
    ok
    user {
      id
      name
      email
    }
    errors
  }
}

Working with arguments in queries and mutations

Argument types

class Query(graphene.ObjectType):
    users = graphene.List(
        User,
        first=graphene.Int(default_value=10),
        skip=graphene.Int(default_value=0),
        search=graphene.String(),
        is_active=graphene.Boolean(default_value=True)
    )
    
    def resolve_users(self, info, first, skip, search=None, is_active=True):
        # Filtering and pagination logic
        users = []  # Fetch from database
        
        if search:
            users = [u for u in users if search.lower() in u.name.lower()]
        
        if is_active is not None:
            users = [u for u in users if u.is_active == is_active]
        
        return users[skip:skip+first]

Using InputObjectType

class UserInput(graphene.InputObjectType):
    name = graphene.String(required=True)
    email = graphene.String(required=True)
    age = graphene.Int()

class CreateUserWithInput(graphene.Mutation):
    class Arguments:
        user_data = UserInput(required=True)
    
    user = graphene.Field(User)
    
    def mutate(self, info, user_data):
        user = User(**user_data)
        return CreateUserWithInput(user=user)

Resolvers: data handling logic

Standard resolvers

class User(graphene.ObjectType):
    id = graphene.ID()
    name = graphene.String()
    email = graphene.String()
    posts_count = graphene.Int()
    
    def resolve_posts_count(self, info):
        # Count user posts
        return len(getattr(self, 'posts', []))
    
    def resolve_email(self, info):
        # Access control
        user = info.context.get('user')
        if user and user.id == self.id:
            return self.email
        return None  # Hide email from other users

Asynchronous resolvers

import asyncio

class Query(graphene.ObjectType):
    user = graphene.Field(User, id=graphene.ID())
    
    async def resolve_user(self, info, id):
        # Asynchronous data fetching
        await asyncio.sleep(0.1)  # Simulate DB call
        return User(id=id, name="Async user")

Database connectivity

Integration with SQLAlchemy

from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from graphene_sqlalchemy import SQLAlchemyObjectType

Base = declarative_base()

class UserModel(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)
    created_at = Column(DateTime)

class User(SQLAlchemyObjectType):
    class Meta:
        model = UserModel
        interfaces = (graphene.relay.Node,)

class Query(graphene.ObjectType):
    users = graphene.List(User)
    
    def resolve_users(self, info):
        query = User.get_query(info)
        return query.all()

Integration with Django ORM

from django.db import models
from graphene_django import DjangoObjectType

class UserModel(models.Model):
    name = models.CharField(max_length=100)
    email = models.EmailField()
    created_at = models.DateTimeField(auto_now_add=True)

class User(DjangoObjectType):
    class Meta:
        model = UserModel
        fields = ("id", "name", "email", "created_at")
    
    posts_count = graphene.Int()
    
    def resolve_posts_count(self, info):
        return self.posts.count()

Working with nested objects and relationships

Defining relationships

class Post(graphene.ObjectType):
    id = graphene.ID()
    title = graphene.String()
    content = graphene.String()
    author = graphene.Field(lambda: User)
    tags = graphene.List(lambda: Tag)
    
    def resolve_author(self, info):
        return User(id=self.author_id, name="Post author")
    
    def resolve_tags(self, info):
        return [Tag(name="Python"), Tag(name="GraphQL")]

class User(graphene.ObjectType):
    id = graphene.ID()
    name = graphene.String()
    posts = graphene.List(Post)
    
    def resolve_posts(self, info):
        return [Post(id=1, title="First post", author_id=self.id)]

Solving the N+1 query problem

from promise import Promise
from promise.dataloader import DataLoader

class UserLoader(DataLoader):
    def batch_load_fn(self, user_ids):
        # Load all users in a single query
        users = UserModel.objects.filter(id__in=user_ids)
        user_map = {user.id: user for user in users}
        return Promise.resolve([user_map.get(user_id) for user_id in user_ids])

class Post(graphene.ObjectType):
    author = graphene.Field(User)
    
    def resolve_author(self, info):
        return info.context['user_loader'].load(self.author_id)

Using Graphene‑Django

Project setup

# settings.py
INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'graphene_django',
    'myapp',
]

GRAPHENE = {
    'SCHEMA': 'myproject.schema.schema',
    'MIDDLEWARE': [
        'graphene_django.debug.DjangoDebugMiddleware',
    ],
}

Creating the schema

# schema.py
import graphene
from graphene_django import DjangoObjectType
from .models import User, Post

class UserType(DjangoObjectType):
    class Meta:
        model = User
        fields = "__all__"

class PostType(DjangoObjectType):
    class Meta:
        model = Post
        fields = "__all__"

class Query(graphene.ObjectType):
    users = graphene.List(UserType)
    posts = graphene.List(PostType)
    
    def resolve_users(self, info):
        return User.objects.all()
    
    def resolve_posts(self, info):
        return Post.objects.select_related('author').all()

schema = graphene.Schema(query=Query)

Integration with Flask and FastAPI

Flask with GraphQL

from flask import Flask, request, jsonify
from flask_graphql import GraphQLView
import graphene

app = Flask(__name__)

class Query(graphene.ObjectType):
    hello = graphene.String()
    
    def resolve_hello(self, info):
        user = info.context.get('user')
        return f"Hello, {user.name if user else 'guest'}!"

schema = graphene.Schema(query=Query)

def get_context():
    return {'user': getattr(request, 'user', None)}

app.add_url_rule('/graphql', view_func=GraphQLView.as_view(
    'graphql',
    schema=schema,
    graphiql=True,
    get_context=get_context
))

FastAPI with GraphQL

from fastapi import FastAPI
from starlette_graphene3 import GraphQLApp
import graphene

app = FastAPI()

class Query(graphene.ObjectType):
    hello = graphene.String()
    
    def resolve_hello(self, info):
        return "Hello from FastAPI!"

schema = graphene.Schema(query=Query)

app.mount("/graphql", GraphQLApp(schema=schema))

Authorization and authentication

Passing context

from flask import g
from functools import wraps

def login_required(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if not g.user:
            raise Exception("Authentication required")
        return f(*args, **kwargs)
    return decorated_function

class Query(graphene.ObjectType):
    my_profile = graphene.Field(User)
    
    @login_required
    def resolve_my_profile(self, info):
        user = info.context.get('user')
        return User(id=user.id, name=user.name)

Field‑level access control

class User(graphene.ObjectType):
    id = graphene.ID()
    name = graphene.String()
    email = graphene.String()
    
    def resolve_email(self, info):
        current_user = info.context.get('user')
        if current_user and (current_user.id == self.id or current_user.is_admin):
            return self.email
        return None

Error handling and validation

Custom exceptions

class ValidationError(Exception):
    def __init__(self, message, field=None):
        self.message = message
        self.field = field
        super().__init__(self.message)

class CreateUser(graphene.Mutation):
    class Arguments:
        name = graphene.String(required=True)
        email = graphene.String(required=True)
    
    user = graphene.Field(User)
    errors = graphene.List(graphene.String)
    
    def mutate(self, info, name, email):
        errors = []
        
        if len(name) < 2:
            errors.append("Name must contain at least 2 characters")
        
        if '@' not in email:
            errors.append("Invalid email format")
        
        if errors:
            return CreateUser(errors=errors)
        
        try:
            user = User(name=name, email=email)
            return CreateUser(user=user)
        except Exception as e:
            return CreateUser(errors=[str(e)])

Global error handling

class CustomMiddleware:
    def resolve(self, next, root, info, **args):
        try:
            return next(root, info, **args)
        except ValidationError as e:
            return {"error": e.message, "field": e.field}
        except Exception as e:
            return {"error": "Internal server error"}

Testing GraphQL APIs

Testing with pytest

import pytest
import json
from graphene.test import Client

def test_user_query():
    client = Client(schema)
    
    query = '''
    {
        user(id: "1") {
            id
            name
        }
    }
    '''
    
    result = client.execute(query)
    assert result['data']['user']['id'] == "1"
    assert result['data']['user']['name'] == "Test user"

def test_create_user_mutation():
    client = Client(schema)
    
    mutation = '''
    mutation {
        createUser(name: "New User", email: "test@example.com") {
            ok
            user {
                name
                email
            }
        }
    }
    '''
    
    result = client.execute(mutation)
    assert result['data']['createUser']['ok'] == True
    assert result['data']['createUser']['user']['name'] == "New User"

Testing with Flask

import pytest
from flask import Flask
from flask_graphql import GraphQLView

@pytest.fixture
def app():
    app = Flask(__name__)
    app.add_url_rule('/graphql', view_func=GraphQLView.as_view('graphql', schema=schema))
    return app

def test_graphql_endpoint(app):
    client = app.test_client()
    
    query = {
        "query": "{ user(id: \"1\") { name } }"
    }
    
    response = client.post('/graphql', 
                          data=json.dumps(query),
                          content_type='application/json')
    
    assert response.status_code == 200
    data = json.loads(response.data)
    assert 'data' in data

Performance and optimization

Pagination

class Query(graphene.ObjectType):
    users = graphene.List(
        User,
        first=graphene.Int(default_value=10),
        skip=graphene.Int(default_value=0)
    )
    
    def resolve_users(self, info, first, skip):
        return User.objects.all()[skip:skip+first]

Relay‑compatible pagination

from graphene import relay

class User(graphene.ObjectType):
    class Meta:
        interfaces = (relay.Node,)
    
    id = graphene.ID()
    name = graphene.String()

class Query(graphene.ObjectType):
    users = relay.ConnectionField(User)
    
    def resolve_users(self, info, **args):
        return User.objects.all()

Caching

from functools import lru_cache

class Query(graphene.ObjectType):
    users = graphene.List(User)
    
    @lru_cache(maxsize=128)
    def resolve_users(self, info):
        return User.objects.all()

Tools and IDEs for working with GraphQL

GraphiQL

Built‑in interface for testing queries:

# Flask
GraphQLView.as_view('graphql', schema=schema, graphiql=True)

# Django
GraphQLView.as_view(graphiql=True)

Additional tools

• GraphQL Playground – modern alternative to GraphiQL
• Postman – GraphQL request support
• Insomnia – specialized REST/GraphQL client
• Apollo Client DevTools – browser extension
• VSCode GraphQL – extensions for auto‑completion and syntax highlighting

Practical use‑case examples

Unified API for different clients

class Query(graphene.ObjectType):
    # Full data for web client
    user_detailed = graphene.Field(User, id=graphene.ID())
    
    # Minimal data for mobile app
    user_mobile = graphene.Field(User, id=graphene.ID())
    
    def resolve_user_detailed(self, info, id):
        return User.objects.select_related('profile', 'settings').get(id=id)
    
    def resolve_user_mobile(self, info, id):
        return User.objects.only('id', 'name', 'avatar').get(id=id)

Aggregating data from multiple sources

class Query(graphene.ObjectType):
    dashboard = graphene.Field(DashboardType, user_id=graphene.ID())
    
    def resolve_dashboard(self, info, user_id):
        # Data from the database
        user = User.objects.get(id=user_id)
        
        # Data from an external API
        external_data = fetch_external_data(user_id)
        
        # Data from cache
        cached_stats = get_cached_stats(user_id)
        
        return DashboardType(
            user=user,
            external_info=external_data,
            statistics=cached_stats
        )

Complete description of the Graphene library

Main modules and components

Scalar types

Decorators and utilities

Schema methods

Integration packages

Frequently asked questions

What is Graphene and how does it differ from other GraphQL libraries?

Graphene is the most mature and feature‑rich library for building GraphQL APIs in Python. It stands out for its declarative schema definition approach, extensive ecosystem of integrations, and an active developer community.

How does Graphene solve the N+1 query problem?

Graphene supports the DataLoader pattern for batching requests and integrates with ORMs to optimize database access via select_related() and prefetch_related() in Django or eager loading in SQLAlchemy.

Can Graphene be used with asynchronous frameworks?

Yes, Graphene supports asynchronous resolvers and can work with FastAPI, Starlette, and other async frameworks through the appropriate integrations.

How to secure a GraphQL API?

It is recommended to use query depth limits, timeouts, input validation, resolver‑level authentication, and query complexity analysis to prevent DoS attacks.

Does Graphene support GraphQL subscriptions?

Yes, Graphene supports subscriptions via graphene.ObjectType using WebSocket or Server‑Sent Events for real‑time updates.

How to test a GraphQL API built with Graphene?

Graphene provides a built‑in test client graphene.test.Client that allows executing queries without an HTTP server. Standard testing tools of the underlying frameworks can also be used.

Can Graphene be used to build a microservice architecture?

Yes, Graphene is suitable for microservices. You can create a GraphQL gateway that aggregates data from multiple services or use GraphQL Federation to combine schemas.

How to migrate from a REST API to GraphQL using Graphene?

Migrations can be incremental: first create GraphQL endpoints alongside REST, then gradually move clients to GraphQL. Graphene makes it easy to wrap existing business logic in GraphQL resolvers.

Full reference of Graphene methods and functions

Graphene provides a powerful and flexible toolkit for building modern GraphQL APIs in Python, supporting all major GraphQL specification features and offering convenient integrations with popular web frameworks and ORM systems.