С недавнего времени в Starlette прекращена поддержка GraphQL. Так что если вы, как и мы, занимались разработкой сервиса на FastAPI, то обновления до последней версии Starlette вас неприятно удивили.

Причины, по которым это случилось, не столь важны, остается просто принять произошедшее как данность. Но переходить с GraphQL обратно на REST нам не хотелось, стандарт подходил под наши задачи, а поэтому надо было найти альтернативу. Всем привет, это Данил Максимов, программист ZeBrains, в этой статье я расскажу, почему после обновления «жить стало лучше, жить стало веселее»(с), и на что надо обратить внимание при миграции на альтернативное решение.

Выбор библиотеки: почему мы остановились на Ariadne

Самый простой вариант выглядел очевидным: поддержка GraphQL в Starlette изначально была построена на базе Graphene, а в качестве одной из альтернатив предлагалась библиотека starlette-graphene3. Но мы уже успели «оценить» и слишком лаконичную документацию, и отсутствие стандартов для написания кода, и проблемы с расширяемостью. 

А потому свой выбор мы остановили на Ariadne: 

• У него достаточно объемная и понятная документация.

• Он построен на базе Apollo Federation, что позволяет пользоваться всеми плюшками от него.

• Активно развивается и не является форками или доработками чужих решений — это самостоятельный продукт.

Главное, что привлекло наше внимание — в отличие от Graphene Ariadne идет от «обратного». Основа — graphql-схема, а по ней строятся все запросы, мутации или подписки.

Это дает гораздо больше гибкости для работы с типами, а также позволяет выстроить систему ошибок и описать ее в схеме. В варианте с использованием решения от Graphene набор ошибок, которые может вернуть мутация, никак не отражен в схеме (только если вы не патчите вручную методы ее построения), а значит — непредсказуем, если у вас нет исходного кода или документации. 

Кроме того, Ariadne прекрасно поддерживает на уровне библиотеки механизм подписок, и для этого не придется тянуть лишние либы (в отличие от того же Graphene). Плюс поддержка передачи файлов при работе с GraphQL изначально была не самой простой задачей, а Ariadne предоставляет ее «из коробки».

Реализация мутаций, запросов и подписок

Писать очередную статью «как переехать с библиотеки ХХХ на библиотеку YYY» — неинтересно. Если вы дочитали до этого момента, значит — в состоянии самостоятельно установить нужные зависимости. Мы же поговорим о том, на что стоит обратить внимание после переезда, что изменится непосредственно в коде. Рассматривать будем, как водится, на примере классического todo-приложения, демо-версия доступна по ссылке.

Запросы и мутации

В GraphQL запросы обрабатываются с помощью резолверов (преобразователей), каждый из которых принимает в себя два позиционных аргумента: obj и info. Пример из документации Ariadne:

def example_resolver(obj: Any, info: GraphQLResolveInfo): 
  return obj.do_something() 

class FormResolver: 
  def __call__(self, obj: Any, info: GraphQLResolveInfo, **data):
    . . .

Из кода выше мы видим, что нам доступны как функциональный, так и ООП подход. Первым делом — определим тип запросов Query в .graphql:

queries/schema.graphql

…

type Query {
    getTasks(userId: ID!): [TaskType]!
    getTask(userId: ID!, taskId: ID!): TaskType!
}

…

Привяжем резолвер к допустимому типу поля схемы с помощью ObjectType, для которого необходимо будет указать метод .set_field(). Он принимает в себя два параметра: name, которое связывает его с одноименным полем схемы GraphQL и, собственно, нужный нам резолвер.

queries/__init__.py

from ariadne import ObjectType

from ariadne_example.app.api.queries import task

queries = ObjectType("Query")

queries.set_field("getTasks", task.resolve_get_user_tasks)
queries.set_field("getTask", task.resolve_get_user_task_by_id)

Сами резолверы импортируются из отдельного файла, давайте их напишем:

queries/task.py

import json
from typing import Any, List

from ariadne import convert_kwargs_to_snake_case
from graphql import GraphQLResolveInfo
from graphql_relay.node.node import from_global_id
from sqlmodel import select

from ariadne_example.app.db.session import Session, engine
from ariadne_example.app.models import Task

@convert_kwargs_to_snake_case
def resolve_get_user_tasks(
        obj: Any,
        info: GraphQLResolveInfo,
        user_id: str,
) -> List[dict]:
    """Get user tasks"""
    with Session(engine) as session:
        local_user_id, _ = from_global_id(user_id)
        statement = select(Task).where(Task.user_id == int(local_user_id))
        tasks = session.execute(statement).scalars().all()
        return [
            Task(
                id=task.id,
                created_at=task.created_at,
                title=task.title,
                status=task.status,
                user_id=task.user_id
            ).dict()
            for task in tasks
        ]

@convert_kwargs_to_snake_case
def resolve_get_user_task_by_id(
        obj: Any,
        info: GraphQLResolveInfo,
        task_id: str,
        user_id: str,
) -> dict:
    """Get user task by task ID."""
    with Session(engine) as session:
        local_task_id, _ = from_global_id(task_id)
        local_user_id, _ = from_global_id(user_id)
        statement = select(Task).where(Task.user_id == local_user_id, Task.id == local_task_id)
        task = session.execute(statement).scalar_one()
        return Task(
            id=task.id,
            created_at=task.created_at,
            title=task.title,
            status=task.status,
            user_id=task.user_id,
        ).dict()

Стандартный для todo-приложения набор резолверов, с помощью которого мы получаем список всех задач пользователя или какую-то конкретную задачу. 

Чуть сложнее ситуация обстоит с мутациями. Поскольку в Ariadne основой всего является схема .graphql, добавим в нее тип, соответствующий нашим мутациям:

schema.graphql

. . .

type Mutations {
    createTask(userId: ID!, taskInput: TaskInput): Response
    changeTaskStatus(taskId: ID!, newSatus: TaskStatusEnum): Response
}

. . .

Для обработки схемы нам потребуется резолвер, который мы сопоставим с мутацией.

mutations/__init__.py

from ariadne import ObjectType

from .task import resolve_create_task

mutations = ObjectType('Mutation')

mutations.set_field('createTask', resolve_create_task)

В коде выше мы импортировали резолвер из файла task.py, давайте его напишем. 

Резолверы мутаций в Ariadne — функции, которые принимают в себя аргументы parent и info, а также произвольный набор аргументов, относящихся к мутации, и возвращают данные, которые отправляются пользователю как результат запроса. В нашем примере описаны два резолвера, отвечающие за создание новой задачи и изменение статуса уже существующей:

mutations/task.py

from typing import Any

import sqlalchemy.exc
from ariadne import convert_kwargs_to_snake_case
from graphql.type.definition import GraphQLResolveInfo
from graphql_relay.node.node import from_global_id
from sqlmodel import select

from ariadne_example.app.db.session import Session, engine
from ariadne_example.app.core.struсtures import TaskStatusEnum, TASK_QUEUES
from ariadne_example.app.models import Task
from ariadne_example.app.core.exceptions import NotFoundError

@convert_kwargs_to_snake_case
def resolve_create_task(
        obj: Any,
        info: GraphQLResolveInfo,
        user_id: str,
        task_input: dict,
) -> int:
    with Session(engine) as session:
        local_user_id, _ = from_global_id(user_id)
        try:
            task = Task(
                title=task_input.get("title"),
                created_at=task_input.get("created_at"),
                status=task_input.get("status"),
                user_id=local_user_id
            )
            session.add(task)
            session.commit()
            session.refresh(task)
        except sqlalchemy.exc.IntegrityError:
            raise NotFoundError(msg='Не найден пользователь с таким user_id')
        return task.id


@convert_kwargs_to_snake_case
async def resolve_change_task_status(
        obj: Any,
        info: GraphQLResolveInfo,
        new_status: TaskStatusEnum,
        task_id: str,
) -> None:
    with Session(engine) as session:
        local_task_id, _ = from_global_id(task_id)
        try:
            statement = select(Task).where(Task.id == local_task_id)
            task = session.execute(statement)
            task.status = new_status
            session.add(task)
            session.commit()
            session.refresh(task)
        except sqlalchemy.exc.IntegrityError:
            raise NotFoundError(msg='Не найдена задача с таким task_id')
        for queue in TASK_QUEUES:
            queue.put(task)

Тут важно обратить внимание, что полезная нагрузка, которую возвращает мутация, представлена в виде простого dict. У нас нет возможности реализовать, как в Graphene класс, и указать в нем ожидаемые поля:

(вариант graphene)task.py

. . .

class CreateTask(graphene.Mutation):
    task = graphene.Field(Task)

    class Arguments:
        user_id = graphene.ID()
        input_data = graphene.Field()

    def mutate(self, parent, info, user_id: str, input_data: Task):
        local_user_id, _ = from_global_id(user_id)
        session = get_session()
        task = Task(title=input_data.get("title"), user_id=local_user_id)
        session.add(task)
        session.commit()
        session.refresh()
        return CreateTask(task=task)


. . .

Но прежде чем приступить к решению этой проблемы, давайте разберемся с третьим типом операции — с подписками.

Подписки в Ariadne

По устоявшейся традиции, первым делом определим тип в схеме:

schema.graphql

. . .

type Subscription {
    taskStatusChanged: TaskType!
}

. . .

Подписки сложнее запросов, поскольку работают не для одиночного обращения к серверу, а должны позволять уведомлять клиента при каждом изменении данных. 

Реализовывать это мы будем «по классике», с использованием WebSockets. Но просто открыть сокет — мало, нам понадобится генератор, который будет передавать данные при их изменении. Кроме того, не помешает иметь и «приемник», в который эти данные будут поступать.

subscriptions.py

import asyncio
from typing import Any

from ariadne import convert_kwargs_to_snake_case, SubscriptionType
from graphql import GraphQLResolveInfo

from ariadne_example.app.core.struсtures import TASK_QUEUES
from ariadne_example.app.models import Task

subscription = SubscriptionType()


@subscription.source("taskStatusChanged")
@convert_kwargs_to_snake_case
async def task_source(obj: Any, info: GraphQLResolveInfo):
    queue = asyncio.Queue()
    TASK_QUEUES.append(queue)
    try:
        while True:
            change_task = await queue.get()
            queue.task_done()
            yield change_task
    except asyncio.CancelledError:
        TASK_QUEUES.remove(queue)
        raise


@subscription.field("taskStatusChanged")
@convert_kwargs_to_snake_case
def task_resolver(task: Task, info: Any):
    return task

Источник подписки мы указываем в subscription.source("taskStatusChanged"), генератор открывает сокет и транслирует нужные нам данные, а резолвер принимает их и передает пользователю.

Ошибки, эксепшены и мидлвары

Все, изложенное выше — сродни обычному тестовому заданию «напишите todo с использованием следующих технологий…». А теперь — поговорим серьезно :-)

Пункт первый — мы отложили «на сладкое» вопрос формализации полезной нагрузки в мутациях. Пункт второй — классический слой ошибок GraphQL в целом позволяет прокинуть код ошибки в extensions и потом его оттуда получать, но для фронта было проблемой определить,  какой именно запрос или мутация завершился с ошибкой и как на эти ошибки реагировать.

Вспомним, что Ariadne пропагандирует подход «от схемы к коду», и добавим в .graphql нужные нам типы ошибок и статусов задач:

schema.graphql

. . .

enum ErrorTypeEnum {
    SERVER_ERROR
    NOT_FOUND_ERROR
    VALIDATION_ERROR
}

type ErrorType {
    message: String
    code: ErrorTypeEnum!
    text: String
}

enum TaskStatusEnum {
    draft
    in_process
    delete
    done
}

. . .

Вынесем логику обработки в core и зададим структуру:

core/structures.py

import enum
from typing import Optional
from dataclasses import dataclass

from ariadne import EnumType, ScalarType


class ErrorTypes(enum.Enum):
    SERVER_ERROR = enum.auto()
    NOT_FOUND_ERROR = enum.auto()
    VALIDATION_ERROR = enum.auto()


@dataclass
class ErrorScalar:
    message: Optional[str]
    code: ErrorTypes
    text: Optional[str]


class TaskStatusEnum(enum.Enum):
    draft = "draft"
    in_process = "in_process"
    delete = "delete"
    done = "done"

task_type_enum = EnumType("TaskStatusEnum", TaskStatusEnum)
datetime_scalar = ScalarType("DateTime")

@datetime_scalar.serializer
def serialize_datetime(value):
    return value.isoformat()

TASK_QUEUES = []

Теперь у нас есть класс, возвращающий осмысленный код ошибки, сообщение и опциональный текст, список вариантов ошибок и статусов задачи. Импортируем их в файл эксепшенов:

core/exceptions.py

from typing import Optional, Dict, Any

from graphql import GraphQLError

from ariadne_example.app.core.struсtures import ErrorTypes, ErrorScalar


class BaseGraphQLError(GraphQLError):
    def __init__(self, msg: str = "Server Error", extensions: Optional[Dict[str, Any]] = None):
        if not hasattr(self, "_extensions"):
            self._extensions = {"code": ErrorTypes.SERVER_ERROR.name}
        if extensions is not None:
            self._extensions = {**self._extensions, **extensions}
        super().__init__(msg, extensions=self._extensions)

    def parse(self) -> ErrorScalar:
        parsed_exception = ErrorScalar(
            message=self.extensions.get("user_message"),
            code=self.extensions.get("code"),
            text=self.message,
        )
        return parsed_exception


class ValidationError(BaseGraphQLError):
    def __init__(self, msg: str, extensions: Optional[Dict[str, Any]] = None):
        self._extensions = {"code": ErrorTypes.VALIDATION_ERROR.name}
        super().__init__(msg, extensions=extensions)


class NotFoundError(BaseGraphQLError):
    def __init__(self, msg: str, extensions: Optional[Dict[str, Any]] = None):
        self._extensions = {"code": ErrorTypes.NOT_FOUND_ERROR.name}
        super().__init__(msg, extensions=extensions)

Теперь мы можем вернуть не просто ошибку GraphQL или авторизации, а конкретный, причем формализованный тип нашей ошибки и ее код. Но зачем останавливаться на достигнутом? Вспомним о middlewares, которые позволят нам обработать написанные шагом ранее эксепшены:

core/middlewares.py

from ariadne.contrib.tracing.utils import is_introspection_field

from ariadne_example.app.core.exceptions import BaseGraphQLError


async def handle_error_middleware(resolver, obj, info, **args):
    """
    Если на этапе выполнения мутации или запроса будет выброшено исключение,
    перехватить и вывести в качестве ошибки.
    """
    errors = []
    value = {}

    if is_introspection_field(info):
        return resolver(obj, info, **args)

    try:
        value = await resolver(obj, info, **args)
    except BaseGraphQLError as exc:
        errors.append(exc.parse())
        value = {**value, **{'errors': errors}}
    except TypeError:
        value = resolver(obj, info, **args)
    return value

Краткий итог

Узнать о том, что разработчик библиотеки отказался от поддержки нужного вам функционала — конечно, неприятно. Но это шанс сделать все так, как хочется именно вам. 

Нам не хватало системы ошибок, чтобы фронт мог связать каждую с конкретным запросом или мутацией. Благодаря (пусть и вынужденному) переходу на Ariadne — мы получили возможность гибко настраивать схему GraphQL под свои задачи, через мидлвары автоматически перехватывать нужные исключения и форматировать их для работы с ошибками на фронте. А самое главное — у нас заработали подписки!

По ссылке — оба варианта реализации тестового приложения todo: на базе  Starlette, и на базе Ariadne. Вопросы, пожелания и проклятия — можно постить в комментариях или ко мне в телеграм @maximovd