Пишете на Python и давно хотели запрыгнуть на поезд хайпа по GraphQL, но никак не могли выбрать между Graphene и Ariadne? Предлагаем вам третий вариант – Strawberry.

Strawberry – code-first библиотека с большим количеством батареек. 2.6 тыс. звёзд в репозитории на GitHub. Для описания типов можно использовать dataclasses и pydantic-модели. Из коробки поддерживается асинхронность.

В этом гайде мы напишем приложение, реализующее создание и получение пользователей (users) и их книг (books).

  • Добавить пользователя (user).
  • Добавить книгу (book) со ссылкой на пользователя.
  • Получить книги со вложенными в них пользователями.

Хранить сущности будем в базе данных, значит, нашим приложением будет IO-bound. Поэтому будем писать асинхронный код.

Для реализации приложения:

  • воспользуемся библиотекой Strawberry с движком fastAPI на Python 3.10;
  • в качестве базы будем использовать postgreSQL версии 14 с python-библиотекой encode/databases и движком asyncpg;
  • для моделей воспользуемся pydantic;
  • напишем запросы с курсорной relay-style пагинацией и dataloader'ами;
  • напишем мутации;
  • напишем тесты с pytest.

Подразумеваем, что вы уже знакомы с концепциями GraphQL. Изучить их можно на официальном портале GraphQL.

Подготовка базы данных

Для запуска и тестирования приложения понадобится запущенная СУБД postgreSQL и созданная в ней база данных.

docker run -d --restart=always -p 5432:5432 -e POSTGRES_PASSWORD=postgres -e POSTGRES_USER=postgres --name postgres postgres:14.3-alpine
psql "postgresql://postgres:postgres@localhost:5432" -с 'create database strawberry;'

Зависимости

Первым делом установим зависимости. Для этого воспользуемся менеджером зависимостей Poetry. Установим его:

pip install poetry

Инициализируем приложение.

poetry init

Всё по умолчанию, кроме вопросов про зависимости (dependencies). Там отвечаем no.

Ставим пакеты:

  • Strawberry – сама GraphQL-библиотека. Устанавливаем версию с интеграцией с fastAPI;
  • Databases – библиотека для работы с базами. Устанавливаем версию с драйвером asyncpg;
  • Uvicorn – асинхронный сервер для запуска приложения;
  • Yoyo-migrations – утилита для миграций;
  • Psycopg2-binary – движок, необходимый для проведения миграций;
  • Pypika – простенький SQL query builder.

И пакеты для разработки:

  • pytest для тестов;
  • pytest-asyncio для асинхронных тестов;
  • pytest-mock для удобства работы с mock;
  • mypy для проверки типов;
  • httpx для тестирования реальными запросами.
poetry add 'strawberry-graphql[fastapi]@0.128.0' 'databases[asyncpg]' 'uvicorn[standard]' yoyo-migrations psycopg2-binary pypika
poetry add -D pytest pytest-asyncio pytest-mock mypy httpx

Структура приложения

Создадим сразу всю структуру приложения.

mkdir migrations
touch migrations/000001.init.sql
touch migrations/000001.init.rollback.sql
touch run.py
touch settings.py
mkdir -p src/users
mkdir -p src/books
touch src/__init__.py
touch src/context.py
touch src/users/__init__.py
touch src/users/gql.py
touch src/users/models.py
touch src/books/__init__.py
touch src/books/gql.py
touch src/books/models.py
mkdir -p tests/fixtures
touch tests/__init__.py
touch tests/conftest.py
touch tests/test_users.py
touch tests/test_books.py
touch tests/fixtures/__init__.py
touch tests/fixtures/clients.py
touch tests/fixtures/graphql_client.py
touch .env
touch mypy.ini

Минимально рабочее приложение

В .env добавим порт запуска приложения и параметры подключения к базе:

PORT=8002
POSTGRES_USER=postgres
POSTGRES_PASSWORD=postgres
POSTGRES_HOST=localhost
POSTGRES_PORT=6432
POSTGRES_DB_NAME=postgres

Будем читать из .env в settings.py:

import os

PORT = int(os.environ.get('PORT', 8001))
DB_USER = os.environ['POSTGRES_USER']
DB_PASSWORD = os.environ['POSTGRES_PASSWORD']
DB_SERVER = os.environ['POSTGRES_HOST']
DB_PORT = int(os.environ['POSTGRES_PORT'])
DB_NAME = os.environ['POSTGRES_DB_NAME']
CONN_TEMPLATE = (
    'postgresql+asyncpg://{user}:{password}@{host}:{port}/{name}'
)
MIGRATIONS_CONN_TEMPLATE = (
    'postgresql://{user}:{password}@{host}:{port}/{name}'
)
DEFAULT_LIMIT = 100

Опишем минимальную версию приложения с подключением к базе в run.py:

import functools as fn
import typing as tp

import databases
import settings
import strawberry
import uvicorn
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter
from strawberry.schema.config import StrawberryConfig


@strawberry.type
class Query():
    """Query."""

    @strawberry.field
    def hello(self) -> str:
        return 'world'


schema = strawberry.Schema(
    query=Query,
    config=StrawberryConfig(auto_camel_case=True),
)


async def startup_db(db: databases.Database):
    await db.connect()


async def shutdown_db(db: databases.Database):
    await db.disconnect()


HOOK_TYPE = tp.Optional[
    tp.Sequence[tp.Callable[[], tp.Any]]
]


def get_app(
    db: databases.Database,
    on_startup: HOOK_TYPE = None,
    on_shutdown: HOOK_TYPE = None,
) -> FastAPI:
    app = FastAPI(
        on_startup=[fn.partial(startup_db, db)]
        if on_startup is None else on_startup,
        on_shutdown=[fn.partial(shutdown_db, db)]
        if on_shutdown is None else on_shutdown,
    )
    graphql_app = GraphQLRouter(
        schema,
    )
    app.include_router(graphql_app, prefix='/graphql')
    return app


def main() -> None:
    database = databases.Database(
        settings.CONN_TEMPLATE.format(
            user=settings.DB_USER,
            password=settings.DB_PASSWORD,
            port=settings.DB_PORT,
            host=settings.DB_SERVER,
            name=settings.DB_NAME,
        ),
    )
    app = get_app(database)
    uvicorn.run(
        app,
        host='0.0.0.0',
        port=settings.PORT,
    )


if __name__ == '__main__':
    main()

Запустим:

poetry run python3 run.py

И откроем в браузере: http://localhost:8002/graphql.

Откроется GraphiQL Playground. В поле для запроса дёрнем нашу единственную ручку:

{
  hello
}

В результате выполнения получим:

{
  "data": {
    "hello": "world"
  }
}

Модели

Миграция – migrations/000001.init.sql:

create table users (
    id bigserial primary key,
    name text not null
);
create table books (
    id bigserial primary key,
    user_id bigint not null references users(id) on delete cascade,
    title text not null
);

Обратная миграция – migrations/000001.init.rollback.sql:

drop table books;
drop table users;

Опишем модели, их создание и получение.

Для описания моделей и структур для создания воспользуемся pydantic-моделями.

Пользователи – src/users/models.py:

from typing import Optional
from databases import Database
import pydantic
from pypika.dialects import PostgreSQLQuery as Query, Table


class User(pydantic.BaseModel):
    id: int
    name: str


class CreateUserInput(pydantic.BaseModel):
    name: str


async def get_users(
    db: Database,
    ids: Optional[list[int]] = None,
) -> list[User]:
    users_tb = Table('users')
    query = Query.from_(users_tb).select(users_tb.star)

    if ids is not None:
        if not ids:
            return []
        query = query.where(
            users_tb.field('id').isin(ids),
        )

    return [
        User(**el._mapping)
        for el in await db.fetch_all(query=str(query))
    ]


async def create_user(
    db: Database,
    create_input: CreateUserInput,
) -> User:
    users_tb = Table('users')
    query = Query.into(users_tb).columns(
        tuple(create_input.dict().keys()),
    ).returning(users_tb.star)
    for value in create_input.dict().values():
        query = query.insert(value)
    row = await db.fetch_one(query=str(query))
    if not row:
        raise ValueError('empty row returned')
    return User(**row._mapping)

Книги – src/books/models.py:

from typing import Optional
from databases import Database
import pydantic
from pypika.dialects import PostgreSQLQuery as Query, Table


class Book(pydantic.BaseModel):
    id: int
    title: str
    user_id: int


class CreateBookInput(pydantic.BaseModel):
    title: str
    user_id: int


async def get_books(
    db: Database,
    ids: Optional[list[int]] = None,
    after: str | None = None,
    first: int | None = None,
) -> list[Book]:
    books_tb = Table('books')
    query = Query.from_(books_tb).select(books_tb.star)

    if ids is not None:
        if not ids:
            return []
        query = query.where(
            books_tb.field('id').isin(ids),
        )
    if after:
        query = query.where(
            books_tb.field('id').gt(int(after)),
        )
    if first:
        query = query.limit(first)

    return [
        Book(**el._mapping)
        for el in await db.fetch_all(query=str(query))
    ]


async def create_book(
    db: Database,
    create_input: CreateBookInput,
) -> Book:
    books_tb = Table('books')
    query = Query.into(books_tb).columns(
        tuple(create_input.dict().keys()),
    ).returning(books_tb.star).insert(
        list(create_input.dict().values()),
    )
    row = await db.fetch_one(query=str(query))
    if not row:
        raise ValueError('empty row returned')
    return Book(**row._mapping)

В get_books мы заранее описали, помимо фильтра по айдишникам, курсорную пагинацию:

  • first - аналог limit в limit-offset’ной пагинации;
  • after - курсор; здесь будем считать, что курсор - айдишник сущности.

Типы для пользователей

Опишем тип пользователя – UserType.

Для этого воспользуемся конвертацией pydantic-типов в типы strawberry. strawberry.auto указывает на то, что мы:

  • копируем поля из модели, указанной в качестве параметра model декоратора type;
  • тип определяется фреймворком автоматически;
  • GraphQL-имя также назначается автоматически в конвенции camelCase.

Сразу же опишем получение юзеров. Для этого опишем тип UserQuery. Параметр extend=True указывает на то, что в переводе на язык GraphQL получится extend type Query.

В этом типе мы создаём resolver с именем user. Для этого нужно описать функцию user и повесить на неё декоратор field. Он принимает в себя параметр id типа int – это айдишник пользователя.

Info[Context, None] – это внутрянка Strawberry. Про него и про user_loader – позже.

Аналогично типу юзера мы опишем структуру создания. Отличие – вместо type используем декоратор input.

Описание структуры создания было нужно, чтобы описать мутацию создания. Очень похоже на описание resolver’а user, но с отличием, что декоратор теперь - mutation.

Обратите внимание на обёртку Annotated. Здесь она позволяет поменять название параметра мутации на input вместо автоматической конвертации create_input_gql в createInputGql.

Ещё один важный момент в мутации – преобразование типа strawberry в pydantic при помощи create_input = create_input_gql.to_pydantic(). Это прогоняет pydantic валидацию и «успокаивает» mypy. :)

Получившийся код в src/users/gql.py:

from typing import Annotated, cast
import strawberry
from src.context import Context
from src.users.models import (
    CreateUserInput,
    User,
    create_user,
)
from strawberry.types import Info



@strawberry.experimental.pydantic.type(model=User, name='User')
class UserType():
    id: strawberry.auto
    name: strawberry.auto


@strawberry.type(name='Query', extend=True)
class UserQuery:

    @strawberry.field(name='user')
    async def user(
        self,
        info: Info[Context, None],
        id: int,
    ) -> UserType | None:
        return await info.context.user_loader.load(id)


@strawberry.experimental.pydantic.input(
    model=CreateUserInput, name='CreateUserInput',
)
class CreateUserInputType:
    name: strawberry.auto


@strawberry.type(name='Mutation', extend=True)
class UserMutation:

    @strawberry.mutation()
    async def create_user(
        self,
        info: Info[Context, None],
        create_input_gql: Annotated[
            CreateUserInputType,
            strawberry.argument(name='input'),
        ],
    ) -> UserType:
        create_input = create_input_gql.to_pydantic()
        return cast(
            UserType,
            await create_user(
                db=info.context.db,
                create_input=create_input,
            ),
        )

Вернёмся к обещанному Info, Context и user_loader, заполним файл src/context.py.

Context можно воспринимать как мешок с полезными вещами – внедрение зависимостей. В GraphQL туда напрашиваются dataloader’ы. Туда их и положим, а также подключение к базе.

Опишем dataloader пользователей. Для этого напишем partial-функцию. Это позволит нам в начале запроса подсунуть в функцию подключение к базе. Так, при вызове load функции dataloader’а, нам останется подсунуть айдишник пользователя.

В src/context.py получилось:

from functools import partial
from typing import TYPE_CHECKING, Optional
import databases
from strawberry.dataloader import DataLoader
from strawberry.fastapi import BaseContext

from src.users.models import get_users

if TYPE_CHECKING:
    from src.users.gql import UserType


class Context(BaseContext):
    """Context."""

    db: databases.Database

    def __init__(
        self,
        db: databases.Database,
        user_loader: DataLoader[int, Optional['UserType']],
    ):
        self.db = db
        self.user_loader = user_loader


def get_context(db: databases.Database) -> Context:
    return Context(
        db=db,
        user_loader=DataLoader(
            load_fn=partial(get_users, db),
        )
    )

Чтобы наше приложение узнало о контексте (и начало его заполнять), а также мутации и получении юзера, модифицируем наш run.py.

В graphql_app появится context_getter:

graphql_app = GraphQLRouter(
    schema,
    context_getter=fn.partial(get_context, db),
)

Из Query пропадёт hello, но отнаследуется UserQuery:

@strawberry.type
class Query(
    UserQuery,
):
    """Query."""

Появятся мутации:

@strawberry.type
class Mutation(
    UserMutation,
):
    """Mutations."""

Мутации надо указать в схеме:

schema = strawberry.Schema(
    query=Query,
    mutation=Mutation,
    config=StrawberryConfig(auto_camel_case=True),
)

Итоговый файл:

import functools as fn
import typing as tp

import databases
import settings
import strawberry
import uvicorn
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter
from strawberry.schema.config import StrawberryConfig

from src.context import get_context
from src.users.gql import UserMutation, UserQuery


@strawberry.type
class Query(
    UserQuery,
):
    """Query."""


@strawberry.type
class Mutation(
    UserMutation,
):
    """Mutations."""


schema = strawberry.Schema(
    query=Query,
    mutation=Mutation,
    config=StrawberryConfig(auto_camel_case=True),
)


async def startup_db(db: databases.Database):
    await db.connect()


async def shutdown_db(db: databases.Database):
    await db.disconnect()


HOOK_TYPE = tp.Optional[
    tp.Sequence[tp.Callable[[], tp.Any]]
]


def get_app(
    db: databases.Database,
    on_startup: HOOK_TYPE = None,
    on_shutdown: HOOK_TYPE = None,
) -> FastAPI:
    app = FastAPI(
        on_startup=[fn.partial(startup_db, db)]
        if on_startup is None else on_startup,
        on_shutdown=[fn.partial(shutdown_db, db)]
        if on_shutdown is None else on_shutdown,
    )
    graphql_app = GraphQLRouter(
        schema,
        context_getter=fn.partial(get_context, db),
    )
    app.include_router(graphql_app, prefix='/graphql')
    return app


def main() -> None:
    database = databases.Database(
        settings.CONN_TEMPLATE.format(
            user=settings.DB_USER,
            password=settings.DB_PASSWORD,
            port=settings.DB_PORT,
            host=settings.DB_SERVER,
            name=settings.DB_NAME,
        ),
    )
    app = get_app(database)
    uvicorn.run(
        app,
        host='0.0.0.0',
        port=settings.PORT,
    )


if __name__ == '__main__':
    main()

Тестирование пользователей

Можно было бы потыкаться руками, но мы же профессионалы.

Из коробки Strawberry позволяет тестировать схему. Но у нас есть зависимости, поэтому приготовим себе инструментарий.

Для начала опишем клиент для отправки запросов. Воспользуемся базовым классом из Strawberry, но допилим его напильником.

  • Скопируем query из базового класса и переделаем так, чтобы был только асинхронный вариант query.
  • Имплементируем request – он будет стучаться асинхронным httpx-клиентом в сервер (об этом чуть позже).
  • Упростим _decode и _build_body.

tests/fixtures/graphql_client.py:

import json
import typing as tp

import databases
from httpx import AsyncClient
from httpx._types import HeaderTypes, RequestFiles
from strawberry.test import BaseGraphQLTestClient, Response


class GraphQLTestClient(BaseGraphQLTestClient):

    def __init__(self, client: AsyncClient, db: databases.Database):
        self._client = client
        self.db = db

    async def query(
        self,
        query: str,
        variables: tp.Optional[tp.Dict[str, tp.Any]] = None,
        headers: tp.Optional[tp.Dict[str, object]] = None,
        asserts_errors: tp.Optional[bool] = True,
        files: tp.Optional[tp.Dict[str, object]] = None,
    ) -> Response:
        """Modifying query to return only sync."""
        body = self._build_body(query, variables, files)

        resp = await self.request(body, headers, files)
        raw_data = self._decode(resp, type='multipart' if files else 'json')

        response = Response(
            errors=raw_data.get('errors'),
            data=raw_data.get('data'),
            extensions=raw_data.get('extensions'),
        )
        if asserts_errors:
            assert response.errors is None, response.errors

        return response

    async def request(
        self,
        body: dict[str, object],
        headers: tp.Optional[dict[str, object]] = None,
        files: tp.Optional[dict[str, object]] = None,
    ):
        """Implement actual request."""
        return await self._client.post(
            '/graphql/',
            json=None if files else body,
            data=body if files else None,
            files=tp.cast(tp.Optional[RequestFiles], files),
            headers=tp.cast(tp.Optional[HeaderTypes], headers),
            follow_redirects=True,
        )

    def _build_body(
        self,
        query: str,
        variables: tp.Optional[dict[str, tp.Mapping]] = None,  # type:ignore
        files: tp.Optional[dict[str, object]] = None,
    ) -> dict[str, object]:
        """Build body to ignore files."""

        body: dict[str, object] = {'query': query}

        if variables:
            body['variables'] = variables

        if files:
            assert variables is not None
            assert files is not None
            file_map = self._build_multipart_file_map(variables, files)

            body = {
                'operations': json.dumps(body),
                'map': json.dumps(file_map),
            }
        return body

    def _decode(
        self,
        response,
        type: tp.Literal['multipart', 'json'],
    ):
        """Always decode to json."""
        return response.json()

На основе клиента опишем фикстуры:

  • event_loop;
  • prepare_db – прогнать миграции и сделать шаблон базы для быстрого дамп-рестор;
  • recreate_db – восстановление базы из шаблона;
  • client – клиент к нашему серверу.

Чуть подробнее про клиент. Мы используем AsyncClient из httpx и подсовываем в него наше приложение. Это позволит делать обычные http-запросы вместо моков. force_rollback = True позволяет запускать тесты в транзакции. Это работает быстрее восстановления из шаблона, но может работать некорректно из-за вложенных транзакций и разных сайд-эффектов параллельной обработки данных. Если что-то сломалось на уровне базы, попробуйте выставить force_rollback = False.

Итого в tests/fixtures/clients.py:

import asyncio

import databases
import pytest
import pytest_asyncio
import settings
from httpx import AsyncClient
from run import get_app
from tests.fixtures.graphql_client import GraphQLTestClient
from yoyo import get_backend, read_migrations

TEST_DB_NAME = 'test_{name}'.format(name=settings.DB_NAME)


@pytest.fixture(scope='session')
def event_loop():
    return asyncio.new_event_loop()


@pytest_asyncio.fixture(autouse=True, scope='session')
async def prepare_db():
    postgres_db = databases.Database(
        settings.CONN_TEMPLATE.format(
            user=settings.DB_USER,
            password=settings.DB_PASSWORD,
            port=settings.DB_PORT,
            host=settings.DB_SERVER,
            name='postgres',
        ),
    )
    async with postgres_db as create_conn:
        await create_conn.execute(
            'drop database if exists {name};'.format(name=TEST_DB_NAME),
        )
        await create_conn.execute(
            'drop database if exists {name}_template;'.format(
                name=TEST_DB_NAME,
            ),
        )
        await create_conn.execute(
            'create database {name};'.format(name=TEST_DB_NAME),
        )

    backend = get_backend(
        settings.MIGRATIONS_CONN_TEMPLATE.format(
            user=settings.DB_USER,
            password=settings.DB_PASSWORD,
            port=settings.DB_PORT,
            host=settings.DB_SERVER,
            name=TEST_DB_NAME,
        ),
    )
    migrations = read_migrations('./migrations')
    with backend.lock():
        backend.apply_migrations(backend.to_apply(migrations))
    del backend  # connection is not released otherwise

    async with postgres_db as template_conn:
        await template_conn.execute(
            'select pg_terminate_backend(pid) '
            'from pg_stat_activity'
            " where datname = \'{name}\'".format(
                name=TEST_DB_NAME,
            ),
        )
        await template_conn.execute(
            'create database {name}_template template {name};'.format(
                name=TEST_DB_NAME,
            ),
        )

    try:
        yield
    finally:
        async with postgres_db as drop_conn:
            await drop_conn.execute(
                'drop database {name};'.format(name=TEST_DB_NAME),
            )
            await drop_conn.execute(
                'drop database {name}_template;'.format(name=TEST_DB_NAME),
            )


async def recreate_db() -> None:
    postgres_db = databases.Database(
        settings.CONN_TEMPLATE.format(
            user=settings.DB_USER,
            password=settings.DB_PASSWORD,
            port=settings.DB_PORT,
            host=settings.DB_SERVER,
            name='postgres',
        ),
    )
    async with postgres_db as drop_conn:
        await drop_conn.execute(
            'drop database {name} with (FORCE);'.format(
                name=TEST_DB_NAME,
            ),
        )
        await drop_conn.execute(
            'create database {name} template {name}_template;'.format(
                name=TEST_DB_NAME,
            ),
        )


@pytest_asyncio.fixture()
async def client():
    # run in transaction - makes tests faster
    # but it can cause problems in case of concurrency
    force_rollback = True

    database = databases.Database(
        settings.CONN_TEMPLATE.format(
            user=settings.DB_USER,
            password=settings.DB_PASSWORD,
            port=settings.DB_PORT,
            host=settings.DB_SERVER,
            name=TEST_DB_NAME,
        ),
        force_rollback=force_rollback,
    )
    async with database as db:
        async with AsyncClient(
            app=get_app(db=database, on_startup=[], on_shutdown=[]),
            base_url='http://test',
        ) as test_client:
            graphql_client = GraphQLTestClient(test_client, db)
            try:
                yield graphql_client
            except Exception:
                pass
    if not force_rollback:
        await recreate_db()

В финале добавим наши фикстуры в tests/conftest.py:

pytest_plugins = (
    'tests.fixtures.clients',
)

Теперь напишем тест на создание в tests/test_users.py. Он проверит мутацию на создание, а также заодно проверит resolver типа пользователя:

import pytest
from tests.fixtures.clients import GraphQLTestClient

pytestmark = [
    pytest.mark.asyncio,
]


create_user_query = """
    mutation createUser(
        $input: CreateUserInput!
    ) {
        createUser(input: $input) {
            id
        }
    }
"""


async def test_create_user(
    client: GraphQLTestClient,
    mocker,
):
    resp = await client.query(
        query=create_user_query,
        variables={
            'input': {
                'name': 'John Doe',
            },
        },
    )
    assert resp.data is not None
    assert resp.data['createUser'] == {'id': mocker.ANY}

Проверим:

poetry run pytest

Типы для книг

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

В типе, по сравнению с пользователями, появился resolver на пользователей – user. Внутри он вызывает dataloader пользователей, подставив в качестве параметра айдишник пользователя, полученный из root – pydantic-модели Book под капотом типа BookType.

Запрос на получение книг с пагинацией оформим по спецификации пагинации от relay. Эту спецификацию рекомендует и официальный сайт GraphQL.

Реализуем тип PageInfo с описанием страницы и generic-типы Connection, Edge с описанием результатов запроса и ребра данных соответственно. В Connection, в дополнении к edges из спецификации relay добавим nodes. Подсмотрено в GraphQL API GitLab’а. Это бывает удобно.

Параметрами запроса на получение будут after и first, про которые мы ранее писали в моделях.

Итоговый файл в src/books/gql.py:

from typing import Generic, TypeVar, cast

import strawberry
from typing import Annotated
import strawberry
from settings import DEFAULT_LIMIT
from src.context import Context
from src.books.models import (
    CreateBookInput,
    Book,
    create_book,
    get_books,
)
from strawberry.types import Info

from src.users.gql import UserType


@strawberry.type
class PageInfo:
    has_next_page: bool
    has_previous_page: bool
    start_cursor: str | None
    end_cursor: str | None


GenericType = TypeVar('GenericType')


@strawberry.type
class Connection(Generic[GenericType]):
    page_info: PageInfo
    edges: list['Edge[GenericType]']
    nodes: list['GenericType']


@strawberry.type
class Edge(Generic[GenericType]):
    node: GenericType
    cursor: str


@strawberry.experimental.pydantic.type(model=Book, name='Book')
class BookType():
    id: strawberry.auto
    title: strawberry.auto

    @strawberry.field
    async def user(
        self,
        root: Book,
        info: Info[Context, None],
    ) -> UserType:
        user = await info.context.user_loader.load(
            key=root.user_id,
        )
        if not user:
            raise ValueError('user not found')
        return user


@strawberry.type(name='Query', extend=True)
class BookQuery:

    @strawberry.field(name='books')
    async def books(
        self,
        info: Info[Context, None],
        after: str | None = None,
        first: int | None = DEFAULT_LIMIT,
    ) -> Connection[BookType]:
        first = (first if first else DEFAULT_LIMIT) + 1

        edges = [
            Edge(
                cursor=str(book.id),
                node=book,
            )
            for book in cast(
                list[BookType],
                await get_books(
                    db=info.context.db,
                    after=after,
                    first=first,
                ),
            )
        ]
        has_next_page = len(edges) == first
        edges = edges[:-1]

        return Connection(
            page_info=PageInfo(
                has_next_page=has_next_page,
                has_previous_page=after is not None,
                start_cursor=edges[0].cursor if edges else None,
                end_cursor=edges[-1].cursor if edges else None,
            ),
            edges=edges,
            nodes=[edge.node for edge in edges],
        )


@strawberry.experimental.pydantic.input(
    model=CreateBookInput, name='CreateBookInput',
)
class CreateBookInputType:
    title: strawberry.auto
    user_id: strawberry.auto


@strawberry.type(name='Mutation', extend=True)
class BookMutation:

    @strawberry.mutation()
    async def create_book(
        self,
        info: Info[Context, None],
        create_input_gql: Annotated[
            CreateBookInputType,
            strawberry.argument(name='input')
        ],
    ) -> BookType:
        create_input = create_input_gql.to_pydantic()
        return cast(
            BookType,
            await create_book(
                db=info.context.db,
                create_input=create_input,
            ),
        )

Добавим книги в схему. Модифицируем run.py. Теперь выглядит так:

import functools as fn
import typing as tp

import databases
import settings
import strawberry
import uvicorn
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter
from strawberry.schema.config import StrawberryConfig

from src.books.gql import BookMutation, BookQuery
from src.context import get_context
from src.users.gql import UserMutation, UserQuery



@strawberry.type
class Query(
    UserQuery,
    BookQuery,
):
    """Query."""


@strawberry.type
class Mutation(
    BookMutation,
    UserMutation,
):
    """Mutations."""


schema = strawberry.Schema(
    query=Query,
    mutation=Mutation,
    config=StrawberryConfig(auto_camel_case=True),
)


async def startup_db(db: databases.Database):
    await db.connect()


async def shutdown_db(db: databases.Database):
    await db.disconnect()


HOOK_TYPE = tp.Optional[
    tp.Sequence[tp.Callable[[], tp.Any]]
]


def get_app(
    db: databases.Database,
    on_startup: HOOK_TYPE = None,
    on_shutdown: HOOK_TYPE = None,
) -> FastAPI:
    app = FastAPI(
        on_startup=[fn.partial(startup_db, db)]
        if on_startup is None else on_startup,
        on_shutdown=[fn.partial(shutdown_db, db)]
        if on_shutdown is None else on_shutdown,
    )
    graphql_app = GraphQLRouter(
        schema,
        context_getter=fn.partial(get_context, db),
    )
    app.include_router(graphql_app, prefix='/graphql')
    return app


def main() -> None:
    database = databases.Database(
        settings.CONN_TEMPLATE.format(
            user=settings.DB_USER,
            password=settings.DB_PASSWORD,
            port=settings.DB_PORT,
            host=settings.DB_SERVER,
            name=settings.DB_NAME,
        ),
    )
    app = get_app(database)
    uvicorn.run(
        app,
        host='0.0.0.0',
        port=settings.PORT,
    )


if __name__ == '__main__':
    main()

Напишем на всё это дело тесты в tests/test_books.py:

import asyncio
import pytest
from src.books.models import CreateBookInput, create_book
from src.users.models import CreateUserInput, create_user
from tests.fixtures.clients import GraphQLTestClient
from tests.test_users import create_user_query
pytestmark = [
    pytest.mark.asyncio,
]


create_book_query = """
    mutation createBook(
        $input: CreateBookInput!
    ) {
        createBook(input: $input) {
            id
            user {
                id
            }
        }
    }
"""

books_query = """
    query books(
        $after: String
        $first: Int
    ) {
        books(
            after: $after
            first: $first
        ) {
            pageInfo {
                hasNextPage
                hasPreviousPage
                startCursor
                endCursor
            }
            edges {
                cursor
                node {
                    id
                }
            }
            nodes {
                id
            }
        }
    }
"""


async def test_create_book(
    client: GraphQLTestClient,
    mocker,
):
    create_user_resp = await client.query(
        query=create_user_query,
        variables={
            'input': {
                'name': 'Ayn Rand',
            },
        },
    )
    user_id = create_user_resp.data['createUser']['id']  # type: ignore
    resp = await client.query(
        query=create_book_query,
        variables={
            'input': {
                'title': 'Atlas shrugged',
                'userId': user_id,
            },
        },
    )
    assert resp.data is not None
    assert resp.data['createBook'] == {
        'id': mocker.ANY,
        'user': {
            'id': user_id,
        }
    }

async def test_get_books(
    client: GraphQLTestClient,
):
    user_inputs = [
        CreateUserInput(name='Ayn Rand'),
        CreateUserInput(name='Fyodor Dostoevsky'),
        CreateUserInput(name='J.K. Rowling'),
    ]
    user_1, user_2, user_3 = await asyncio.gather(
        *(create_user(
            db=client.db,
            create_input=user_input,
        ) for user_input in user_inputs)
    )
    book_inputs = [
        CreateBookInput(title='Atlas Shrugged', user_id=user_1.id),
        CreateBookInput(title='Anthem', user_id=user_1.id),
        CreateBookInput(title='Idiot', user_id=user_2.id),
        CreateBookInput(title='Demons', user_id=user_2.id),
        CreateBookInput(title='Crime and Punishment', user_id=user_2.id),
        CreateBookInput(
            title='Harry Potter and the Philosopher Stone',
            user_id=user_3.id
        ),
        CreateBookInput(
            title='Harry Potter and the Chamber of Secrets',
            user_id=user_3.id
        ),
        CreateBookInput(
            title='Harry Potter and the Prisoner of Azkaban',
            user_id=user_3.id
        ),
    ]
    _, book_2, book_3, book_4, book_5, *_ = await asyncio.gather(
        *(create_book(
            db=client.db,
            create_input=book_input,
        ) for book_input in book_inputs)
    )
    books_resp = await client.query(
        query=books_query,
        variables={
            'after': str(book_2.id),
            'first': 3,
        },
    )
    assert books_resp.data
    assert books_resp.data['books'] == {
        'pageInfo': {
            'hasNextPage': True,
            'hasPreviousPage': True,
            'startCursor': str(book_3.id),
            'endCursor': str(book_5.id),
        },
        'edges': [
            {
                'cursor': str(book_3.id),
                'node': {'id': book_3.id},
            },
            {
                'cursor': str(book_4.id),
                'node': {'id': book_4.id},
            },
            {
                'cursor': str(book_5.id),
                'node': {'id': book_5.id},
            }
        ],
        'nodes': [
            {'id': book_3.id},
            {'id': book_4.id},
            {'id': book_5.id},
        ],
    }

Финальный код можно посмотреть в нашем открытом репозитории.

Strawberry активно развивается, поэтому документации часто не хватает. Отсутствие документации компенсируется активным сообществом в Discord. Maintainer’ы ежедневно в сети, готовы оперативно помочь по всем вопросам.