Эта статья родилась из опыта использования GraphQL в проекте одного из крупнейших аэропортов РФ. Проект посвящен разработке системы по автоматизации обслуживания рейсов и управлению ресурсами аэропорта в реальном времени (MRMS система).
Проект реализован на базе микросервисной архитектуры, где модель данных аэропорта представлена в виде GraphQL API, а сервер, предоставляющий API, написан на java. Клиентами этого API являются не только web/mobile, но и сервисы на java, golang и python.
Статья написана как пошаговое руководство по созданию своего GraphQL клиента на python с нуля, где автор демонстрирует проблемы, возникающие на этом пути. Использовать реальную GraphQL схему аэропорта не представляется возможным, поэтому для наглядности будем использовать открытую схему github GraphQL API.
Содержание
Что нас ждет
Подготовка
Первый запрос
Синхронный клиент
Асинхронный клиент
Создаем клиент
Работа с запросами
Генерация модели данных
Заключение
Что нас ждет
Вот некоторые из задач, которые мы рассмотрим
как совершить GraphQL запрос и получить ответ от сервера?
асинхронные запросы к серверу;
как управлять своей кодовой базой запросов?
хранение ответов от сервера в виде типизированных классов;
Упоминание новых инструментов будет всегда сопровождаться ссылкой на оф. документацию. Но для комфортного чтения статьи желательно знать основы GraphQL. Так же будет плюсом иметь опыт работы с библиотеками requests, asyncio и aiohttp.
Исходный код из статьи можно найти тут. Код написан на python 3.11 с использованием poetry, типизирован и отформатирован с помощью black.
Подготовка
Для начала подготовимся к работе с github GraphQL API. Во-первых, настроим наше python окружение. Будем использовать poetry и следующую начальную структуру файлов
├── github_graphql_client/ <- Тут будет код нашего клиента
│ └── __init__.py
├── tests/ <- Тут будут тесты
│ └── __init__.py
├── scripts/ <- Тут будут различные скрипты для запуска
├── pyproject.toml <- Файл с настройками проекта
├── README.md
├── .env <- Тут будут всякие sensitive переменные
└── .gitignoreКонфиг pyproject.toml следующий
# `pyproject.toml` file
[tool.poetry]
name = "github_graphql_client"
version = "0.1.0"
description = "Github GraphQL client for Habr."
authors = ["FirstName SecondName <email>"]
readme = "README.md"
packages = [{include = "github_graphql_client"}]
[tool.poetry.dependencies]
python = "^3.11"
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"Во-вторых, Github GraphQL endpoint находится по адресу https://api.github.com/graphql. Чтобы воспользоваться API — нужно получить токен. Подробности про выпуск токена описаны тут: Github - Forming calls with GraphQL. В файле .env перечислим следующие параметры
GITHUB_TOKEN=<YOUR_TOKEN>
GITHUB_GRAPHQL_ENDPOINT=https://api.github.com/graphqlДля работы с файлом .env будем использовать пакет python-dotenv. Добавим его в качестве нашей первой зависимости
$ poetry add python-dotenv
Creating virtualenv blah-blah-py3.11 in /home/blah-blah/pypoetry/virtualenvs
Using version ^1.0.0 for python-dotenv
Updating dependencies
Resolving dependencies... (0.1s)
Package operations: 1 install, 0 updates, 0 removals
• Installing python-dotenv (1.0.0)
Writing lock fileПосле выполнения команды poetry add появится файл poetry.lock, а так же наша зависимость будет добавлена в pyproject.toml
# `pyproject.toml` file
# ...
[tool.poetry.dependencies]
python = "^3.11"
python-dotenv = "^1.0.0"
# ...И, наконец, запустим poetry install для установки всех зависимостей.
Первый запрос
Основное, что должен уметь наш клиент — подключаться к серверу. Обычно GraphQL работает по протоколу HTTP/HTTPS через единственный POST, который ожидает на входе json с полями query: str и variables: dict[str, Any]. Так это работает и в github GraphQL API. Схему github GraphQL API можно посмотреть тут. Для начала спроектируем класс, который будет отвечать непосредственно за соединение с сервером и получение данных. Добавим несколько новых файлов
$ mkdir github_graphql_client/transport
$ touch github_graphql_client/transport/__init__.py github_graphql_client/transport/base.pyВ файле github_graphql_client/transport/base.py определим BaseTransport
# `github_graphql_client/transport/base.py` file
from typing import Any
class BaseTransport:
"""An abstract transport."""
def execute(
self, query: str, variables: dict[str, Any], **kwargs: Any
) -> dict[str, Any]:
"""Execute GraphQL query."""
raise NotImplementedError
def connect(self) -> None:
"""Establish a session with the transport."""
raise NotImplementedError
def close(self) -> None:
"""Close a session."""
raise NotImplementedErrorЛюбой класс типа Transport должен наследоваться от BaseTransport. Для этого необходимо реализовать три метода
connect — метод для открытия соединения с сервером;
close — метод для закрытия соединения;
execute — метод для выполнения запроса query с переменными variables;
Так, как GraphQL запрос есть обычный POST запрос — клиент может быть реализован с помощью пакета requests. Для начала добавим зависимость
$ poetry add requestsВ новый файл github_graphql_client/transport/requests.py добавим следующий код
# `github_graphql_client/transport/requests.py` file
from typing import Any, Optional
import requests as r
from github_graphql_client.transport.base import BaseTransport
class RequestsTransport(BaseTransport):
"""The transport based on requests library."""
session: Optional[r.Session]
def __init__(self, endpoint: str, token: str, **kwargs: Any) -> None:
self.endpoint = endpoint
self.token = token
self.auth_header = {"Authorization": f"Bearer {self.token}"}
self.session = None
def connect(self) -> None:
"""Start a `requests.Session` connection."""
if self.session is None:
self.session = r.Session()
else:
raise Exception("Session already started")
def close(self) -> None:
"""Closing `requests.Session` connection."""
if self.session is not None:
self.session.close()
self.session = None
def execute(
self, query: str, variables: dict[str, Any], **kwargs: Any
) -> dict[str, Any]:
"""Execute GraphQL query."""
if self.session is None:
raise Exception(f"RequestsTransport session not connected")
post_args = {
"headers": self.auth_header,
"json": {"query": query, "variables": variables},
}
post_args["headers"]["Content-Type"] = "application/json"
response = self.session.request("POST", self.endpoint, **post_args)
result = response.json()
return result.get("data")В данной реализации
метод connect создает объект requests.Session если он не был создан ранее;
метод close закрывает соединение для объекта requests.Session;
метод execute отправляет с помощью объекта requests.Session обычный POST запрос;
Давайте проверим, что с помощью этого класса мы уже можем получить данные. Отправим query, который должен вернуть несколько первых завершенных issues из выбранного github репозитория. Для примера, рассмотрим новый проект автора pydantic - FastUI. Хранить запросы будем в отдельной папке
$ mkdir github_graphql_client/queries
$ touch github_graphql_client/queries/__init__.py github_graphql_client/queries/repository.py# `github_graphql_client/queries/repository.py` file
repository_issues_query = """
query {
repository(owner:"pydantic", name:"FastUI") {
issues(last:2, states:CLOSED) {
edges {
node {
title
url
}
}
}
}
}
"""Для удобства добавим скрипт scripts/run.py
# `scripts/run.py` file
import os
import time
from typing import Any
from dotenv import load_dotenv
from github_graphql_client.client.requests_client import RequestsClient
from github_graphql_client.queries.repository import repository_issues_query
load_dotenv() # take environment variables from .env
GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN")
GITHUB_GRAPHQL_ENDPOINT = os.environ.get("GITHUB_GRAPHQL_ENDPOINT")
def main():
transport = RequestsTransport(
endpoint=GITHUB_GRAPHQL_ENDPOINT, token=GITHUB_TOKEN
)
transport.connect()
data = transport.execute(
query=repository_issues_query, variables={},
)
print(data)
transport.close()
if __name__ == "__main__":
main()После запуска скрипта мы получим примерно следующее
$ python3 scripts/run.py
{'repository': {'issues': {'edges': [{'node': {'title': 'More PageEvent Triggers', 'url': 'https://github.com/pydantic/FastUI/issues/104'}}, {'node': {'title': 'TypeError: Interval() takes no arguments', 'url': 'https://github.com/pydantic/FastUI/issues/105'}}]}}}Проект активно живет, поэтому сейчас issues будут другими.
Таймаут
Хорошо, если вы понимаете, сколько должен выполняться ваш запрос. Процитируем одну известную статью
Не ставить таймаут на задачу — это зло. Это значит, что вы не понимаете, что происходит в задаче, как должна работать бизнес-логика.
Добавим таймаут в RequestsTransport
# `github_graphql_client/transport/requests.py` file
...
class RequestsTransport(BaseTransport):
"""The transport based on requests library."""
DEFAULT_TIMEOUT: int = 1
session: Optional[r.Session]
def __init__(self, endpoint: str, token: str, **kwargs: Any) -> None:
...
self.timeout = kwargs.get("timeout", RequestsTransport.DEFAULT_TIMEOUT)
...
def execute(
self, query: str, variables: dict[str, Any], **kwargs: Any
) -> dict[str, Any]:
...
post_args = {
"headers": self.auth_header,
"json": {"query": query, "variables": variables},
"timeout": self.timeout,
}
...Проверим наш код указав очень маленький (для github GraphQL API) таймаут
# `scripts/run.py` file
...
def main():
transport = RequestsTransport(
endpoint=GITHUB_GRAPHQL_ENDPOINT,
token=GITHUB_TOKEN,
timeout=0.0001,
)
...
...Запустив scripts/run.py, мы получим следующее
$ python3 scripts/run.py
...
requests.exceptions.ConnectTimeout: HTTPSConnectionPool(host='api.github.com', port=443): Max retries exceeded with url: /graphql (Caused by ConnectTimeoutError(<urllib3.connection.HTTPSConnection object at 0x7f5510137d50>, 'Connection to api.github.com timed out. (connect timeout=0.0001)'))По аналогии вы можете добавить свои настройки для requests, а мы пойдем дальше.
Синхронный клиент
Создадим клиент, который научим работать с BaseTransport. После выполнения команд
$ mkdir github_graphql_client/client
$ touch github_graphql_client/client/__init__.py github_graphql_client/client/sync_client.pyв файл github_graphql_client/client/sync_client.py добавим следующий код
# `github_graphql_client/client/sync_client.py` file
from typing import Any
from github_graphql_client.transport.base import BaseTransport
class SyncGraphQLClient:
"""Sync GraphQL client based on `BaseTransport` transport."""
transport: BaseTransport
def __init__(self, transport: BaseTransport) -> None:
self.transport = transport
def __enter__(self):
self.connect_sync()
return self
def __exit__(self, *args):
self.close_sync()
def connect_sync(self) -> None:
"""Connect to `self.transport`."""
self.transport.connect()
def close_sync(self) -> None:
"""Close `self.transport` connection."""
self.transport.close()
def execute_sync(
self, query: str, variables: dict[str, Any], **kwargs: Any
) -> dict[str, Any]:
return self.transport.execute(query, variables, **kwargs)Класс SyncGraphQLClient умеет запускать сессию для произвольного класса BaseTransport и получать результаты запросов опуская детали реализации самого запроса.
метод connect_sync создает соединение через self.transport;
метод close_sync закрывает соединение через self.transport;
метод execute_sync получает данные с сервера через self.transport;
методы __enter__ и __exit__ предназначены для того, чтобы запускать клиент в контекстном менеджере и не забывать закрыть соединение;
Проверим, что это работает
# `scripts/run.py` file
...
from github_graphql_client.client.sync_client import SyncGraphQLClient
...
def main():
transport = RequestsTransport(
endpoint=GITHUB_GRAPHQL_ENDPOINT,
token=GITHUB_TOKEN,
)
with SyncGraphQLClient(transport=transport) as client:
data = client.execute_sync(
query=repository_issues_query, variables={},
)
print(data)
...Запустив scripts/run.py, получим тот же результат
$ python3 scripts/run.py
{'repository': {'issues': {'edges': [{'node': {'title': 'More PageEvent Triggers', 'url': 'https://github.com/pydantic/FastUI/issues/104'}}, {'node': {'title': 'TypeError: Interval() takes no arguments', 'url': 'https://github.com/pydantic/FastUI/issues/105'}}]}}}Слово sync в названии SyncGraphQLClient не случайно. Немного позже у нас появится AsyncGraphQLClient, и мы объединим их в один GraphQLClient.
Несколько запросов
Давайте внутри одной сессии выполним несколько GraphQL запросов. Для этого изменим наш запрос так, чтобы иметь возможность указать название желаемого репозитория
# `github_graphql_client/queries/repository.py` file
def get_repository_issues_query(owner: str, name: str) -> str:
return """query {
repository(owner:"%s", name:"%s") {
issues(last:2, states:CLOSED) {
edges {
node {
title
url
}
}
}
}
}
""" % (
owner,
name,
)Наш скрипт scripts/run.py будет выглядеть следующим образом
# `scripts/run.py` file
...
def main():
transport = RequestsTransport(
endpoint=GITHUB_GRAPHQL_ENDPOINT,
token=GITHUB_TOKEN,
)
with SyncGraphQLClient(transport=transport) as client:
data = client.execute_sync(
query=get_repository_issues_query("pydantic", "FastUI"),
variables={},
)
print(data)
data = client.execute_sync(
query=get_repository_issues_query("pydantic", "pydantic"),
variables={},
)
print(data)
data = client.execute_sync(
query=get_repository_issues_query("pydantic", "pydantic-core"),
variables={},
)
print(data)
...После запуска получим примерно следующее
$ python3 scripts/run.py
{'repository': {'issues': {'edges': [{'node': {'title': 'More PageEvent Triggers', 'url': 'https://github.com/pydantic/FastUI/issues/104'}}, {'node': {'title': 'TypeError: Interval() takes no arguments', 'url': 'https://github.com/pydantic/FastUI/issues/105'}}]}}}
{'repository': {'issues': {'edges': [{'node': {'title': "__init__.cpython-311-darwin.so is an incompatible architecture (have 'x86_64', need 'arm64') in M1 mac mini", 'url': 'https://github.com/pydantic/pydantic/issues/8396'}}, {'node': {'title': 'Override class used in annotations', 'url': 'https://github.com/pydantic/pydantic/issues/8408'}}]}}}
{'repository': {'issues': {'edges': [{'node': {'title': '2.14.4 release upload failed', 'url': 'https://github.com/pydantic/pydantic-core/issues/1082'}}, {'node': {'title': "( Интересные статьи
Интересные статьи
