GraphQL в Python: как создать гибкое API с минимальными затратами

# REST подход с Django
@api_view(['GET'])
def get_author_with_books(request, author_id):
author = Author.objects.get(id=author_id)
serializer = AuthorWithBooksSerializer(author)
return Response(serializer.data)

# Специальный сериализатор для этого случая
class AuthorWithBooksSerializer(serializers.ModelSerializer):
books = BookSerializer(many=True)
class Meta:
model = Author
fields = ['id', 'name', 'books']

# GraphQL подход с Graphene
class AuthorType(DjangoObjectType):
class Meta:
model = Author
fields = ('id', 'name', 'books')

class BookType(DjangoObjectType):
class Meta:
model = Book
fields = ('id', 'title', 'pages')

class Query(graphene.ObjectType):
author = graphene.Field(AuthorType, id=graphene.ID(required=True))

def resolve_author(self, info, id):
return Author.objects.get(pk=id)

query {
author(id: "1") {
name
books {
title
}
}
}

# Для Graphene
pip install graphene django-graphql-jwt

# Для Ariadne
pip install ariadne uvicorn

# Для Strawberry
pip install strawberry-graphql[debug-server]

# settings.py
INSTALLED_APPS = [
# ...
'graphene_django',
]

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

# urls.py
from django.urls import path
from graphene_django.views import GraphQLView
from django.views.decorators.csrf import csrf_exempt

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

# schema.py
import graphene

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.py
from ariadne import QueryType, make_executable_schema
from ariadne.asgi import GraphQL
from fastapi import FastAPI

# Определение схемы в SDL
type_defs = """
type Query {
hello(name: String = "World"): String!
}
"""

# Создание резолвера
query = QueryType()

@query.field("hello")
async def resolve_hello(_, info, name="World"):
return f"Hello {name}"

# Создание исполняемой схемы
schema = make_executable_schema(type_defs, query)

# Интеграция с FastAPI
app = FastAPI()
app.mount("/graphql", GraphQL(schema, debug=True))

import graphene
from graphene_django import DjangoObjectType
from .models import Book, Author, Publisher

# Определение типов
class AuthorType(DjangoObjectType):
class Meta:
model = Author
fields = ("id", "name", "biography", "books")

class BookType(DjangoObjectType):
class Meta:
model = Book
fields = ("id", "title", "year", "author", "publisher")

class PublisherType(DjangoObjectType):
class Meta:
model = Publisher
fields = ("id", "name", "founded_year", "books")

# Определение запросов (queries)
class Query(graphene.ObjectType):
book = graphene.Field(BookType, id=graphene.ID(required=True))
books = graphene.List(BookType, 
author_id=graphene.ID(),
year=graphene.Int())
authors = graphene.List(AuthorType)

def resolve_book(self, info, id):
return Book.objects.get(pk=id)

def resolve_books(self, info, author_id=None, year=None):
query = Book.objects.all()

if author_id:
query = query.filter(author_id=author_id)
if year:
query = query.filter(year=year)

return query

def resolve_authors(self, info):
return Author.objects.all()

# Определение входного типа для мутации
class BookInput(graphene.InputObjectType):
title = graphene.String(required=True)
year = graphene.Int(required=True)
author_id = graphene.ID(required=True)
publisher_id = graphene.ID(required=True)

# Определение мутаций
class CreateBook(graphene.Mutation):
class Arguments:
book_data = BookInput(required=True)

book = graphene.Field(BookType)

@staticmethod
def mutate(root, info, book_data):
author = Author.objects.get(pk=book_data.author_id)
publisher = Publisher.objects.get(pk=book_data.publisher_id)

book = Book(
title=book_data.title,
year=book_data.year,
author=author,
publisher=publisher
)
book.save()

return CreateBook(book=book)

class Mutation(graphene.ObjectType):
create_book = CreateBook.Field()

# Создание схемы
schema = graphene.Schema(query=Query, mutation=Mutation)

# schema/__init__.py
import graphene
from .book import BookQuery, BookMutation
from .author import AuthorQuery, AuthorMutation

class Query(BookQuery, AuthorQuery, graphene.ObjectType):
pass

class Mutation(BookMutation, AuthorMutation, graphene.ObjectType):
pass

schema = graphene.Schema(query=Query, mutation=Mutation)

from promise import Promise
from promise.dataloader import DataLoader

class AuthorLoader(DataLoader):
def batch_load_fn(self, author_ids):
authors = {str(author.id): author for author in Author.objects.filter(id__in=author_ids)}
return Promise.resolve([authors.get(str(author_id)) for author_id in author_ids])

# В резолвере
def resolve_book(self, info, id):
book = Book.objects.get(pk=id)
author_loader = info.context.author_loader
book.author = author_loader.load(book.author_id)
return book

def test_book_query():
client = Client(schema)
query = '''
query {
book(id: "1") {
title
author {
name
}
}
}
'''
result = client.execute(query)
assert 'errors' not in result
assert result['data']['book']['title'] == "Ожидаемое название"

# schema.graphql
type Author {
id: ID!
name: String!
biography: String
books: [Book!]!
}

type Book {
id: ID!
title: String!
year: Int!
author: Author!
publisher: Publisher!
}

type Publisher {
id: ID!
name: String!
foundedYear: Int
books: [Book!]!
}

input BookInput {
title: String!
year: Int!
authorId: ID!
publisherId: ID!
}

type Query {
book(id: ID!): Book
books(authorId: ID, year: Int): [Book!]!
authors: [Author!]!
}

type Mutation {
createBook(bookData: BookInput!): BookPayload!
}

type BookPayload {
success: Boolean!
book: Book
errors: [String!]
}

# resolvers.py
from ariadne import ObjectType, QueryType, MutationType, make_executable_schema
from db import get_book, get_books, get_authors, create_book
from db.models import Author, Publisher

# Типы резолверов
query = QueryType()
mutation = MutationType()
book = ObjectType("Book")
author = ObjectType("Author")
publisher = ObjectType("Publisher")

# Резолверы для запросов
@query.field("book")
def resolve_book(_, info, id):
return get_book(id)

@query.field("books")
def resolve_books(_, info, authorId=None, year=None):
return get_books(author_id=authorId, year=year)

@query.field("authors")
def resolve_authors(_, info):
return get_authors()

# Резолверы для связей
@book.field("author")
def resolve_book_author(book_obj, info):
return Author.objects.get(id=book_obj.author_id)

@book.field("publisher")
def resolve_book_publisher(book_obj, info):
return Publisher.objects.get(id=book_obj.publisher_id)

@author.field("books")
def resolve_author_books(author_obj, info):
return author_obj.books.all()

# Мутации
@mutation.field("createBook")
def resolve_create_book(_, info, bookData):
try:
book = create_book(
title=bookData["title"],
year=bookData["year"],
author_id=bookData["authorId"],
publisher_id=bookData["publisherId"]
)
return {
"success": True,
"book": book,
"errors": []
}
except Exception as e:
return {
"success": False,
"book": None,
"errors": [str(e)]
}

# Загрузка схемы из файла
with open("schema.graphql") as schema_file:
type_defs = schema_file.read()

# Создание исполняемой схемы
schema = make_executable_schema(
type_defs, 
query, 
mutation, 
book,
author,
publisher
)

# app.py
from fastapi import FastAPI
from ariadne.asgi import GraphQL
from resolvers import schema

app = FastAPI()

# GraphQL эндпоинт
app.mount("/graphql", GraphQL(schema, debug=True))

# Дополнительно: REST-эндпоинт, возвращающий GraphQL результат
@app.get("/books/{book_id}")
async def get_book(book_id: str):
query = """
query GetBook($id: ID!) {
book(id: $id) {
title
year
author {
name
}
}
}
"""
result = await schema.execute(query, variable_values={"id": book_id})
return result.data["book"]

# В schema.graphql добавляем
type Subscription {
bookAdded: Book!
}

# В resolvers.py
from ariadne import SubscriptionType
import asyncio

subscription = SubscriptionType()

@subscription.source("bookAdded")
async def book_added_source(obj, info):
# Имитация потока событий
while True:
await asyncio.sleep(2)
yield {"bookAdded": get_latest_book()}

@subscription.field("bookAdded")
def book_added_resolver(payload, info):
return payload["bookAdded"]

# Добавляем subscription в схему
schema = make_executable_schema(
type_defs, 
query, 
mutation, 
subscription,
# ...
)

# db/models.py
from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker

Base = declarative_base()
engine = create_engine("sqlite:///library.db")
Session = sessionmaker(bind=engine)

class Author(Base):
__tablename__ = "authors"
id = Column(Integer, primary_key=True)
name = Column(String)
biography = Column(String)
books = relationship("Book", back_populates="author")

class Book(Base):
__tablename__ = "books"
id = Column(Integer, primary_key=True)
title = Column(String)
year = Column(Integer)
author_id = Column(Integer, ForeignKey("authors.id"))
publisher_id = Column(Integer, ForeignKey("publishers.id"))
author = relationship("Author", back_populates="books")
publisher = relationship("Publisher", back_populates="books")

class Publisher(Base):
__tablename__ = "publishers"
id = Column(Integer, primary_key=True)
name = Column(String)
founded_year = Column(Integer)
books = relationship("Book", back_populates="publisher")

# Инициализация БД
Base.metadata.create_all(engine)

from ariadne import format_error
from graphql import GraphQLError

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

# В резолвере мутации
@mutation.field("createBook")
def resolve_create_book(_, info, bookData):
if len(bookData["title"]) < 3:
raise ValidationError("Title must be at least 3 characters", field="title")
# ...

# Настройка обработки ошибок
app.mount(
"/graphql", 
GraphQL(
schema, 
debug=True,
error_formatter=format_error
)
)