responderで理解するWebサーバー #7 GraphQL

juri-t

こんにちは、Webエンジニアのjuri-tです。

GraphQL使ってますか？私は正直聞いたことはありましたが、使ってみたことはありませんでした。responderでは標準でサポートしているということなので、今回良い機会なので使ってみました。

GraphQLを教科書的に説明すると、

Facebookが開発したデータを取得・操作するためのクエリ言語

となります。よく比較対象としてあげられているのはRESTful APIです。RESTful APIは言語ではなく設計思想ですが、それが抱えていた問題を解決するために作られた言語のため、言語よりも以下のような特長がよく語られます。

・欲しいデータをクライアントが宣言的に取得できる
・関連するリソースも一度のリクエストで取得できる

欲しいデータをクライアントが宣言的に取得できる

RESTful APIでは、どのリソースに対するアクセスかは明確ですが、その中のどの要素が必要になっているかは不明瞭です。GraphQLではリソースの中の要素まで宣言するため、わかりやすいメリットがあります。また、不要なデータをやりとりする必要もないため、データの転送量の面でも効率が良いです。

関連するリソースも一度のリクエストで取得できる

RESTful APIでは、リレーションがあるデータを取得する際、N+1問題と呼ばれる問題に当たります。これはざっくりいうと一度あるリソースを取得したあと、関連するすべてのリソースに対して都度リクエストする必要があるという問題です。GraphQLの場合はそれも一度のリクエストで取得できます。（N+1問題は、実際はSQLの発行回数で語られることが多いと思いますが、問題の本質は同じです）

それでは早速、見ていきます。

とりあえずチュートリアルをやってみた

公式サイトに書いてあるサンプルを見ると以下のようなコードが紹介されています。GraphQLを実際に使ったことのない身からすると、正直「イミワカラン・・」って思いました。

import graphene

class Query(graphene.ObjectType):
   hello = graphene.String(name=graphene.String(default_value="stranger"))

   def resolve_hello(self, info, name):
       return f"Hello {name}"

schema = graphene.Schema(query=Query)
view = responder.ext.GraphQLView(api=api, schema=schema)

api.add_route("/graph", view)

最小構成で使うときの例ということなんでしょうが、GraphQLで語られているメリットが全く含まれていません（型にvalidationがあるぐらい？）。このサンプル読む限り、「これ、GraphQLでやる必要ないよね？」というのが第一印象でした。

また、これを実行して、/graphにアクセスすると以下のような画面が見えるのですが、これはDeveloper toolのようなもので、実際のアプリケーションでは使わないものだと思います。

GraphQL、 Graphene-Pythonを読む

responderのGraphQLの説明を見ても埒が明かなかったので、内部的に利用しているgrapheneを読むことにしました。しかし、そもそもGraphQLについてちゃんと知らなかったため、ここでも何を言ってるか理解が進まず、一緒にGraphQLのサイトも読むことにしました。

私もこの記事を書いている時点で、初歩の初歩ぐらいしか理解できていませんが、以下については知っておけばこの記事が多少読みやすくなるのではと思います。

・GraphQLには大きく３つ、Query / Mutation / Subscriptionがある
・GraphQLの実装には大きく２つ、Schema First / Code Firstがある 

GraphQLにはQuery / Mutation / Subscriptionがある

Queryとは、データを取得する処理のことです。Mutationは、データを更新する処理のことですが、とりあえず今回は扱いません。Subscriptionはデータが更新されたことを通知する仕組みのようですが、よくわかってないし、今回は扱いません。

GraphQLの実装にはSchema First / Code Firstがある

スキーマ定義（Schema Definition Language）を書くことで実装するスタイルと、スキーマ定義は書かずコードを書くスタイルがあるようです。

grapheneのサイトに書いてある内容になりますが、スキーマファーストで実装されているライブラリは、JavaScriptのAppolo ServerやPythonのAriadneがあり、grapheneはコードファーストの実装となっているようです。

リレーションを貼っているデータ（想定）を取得する

今回の目標はこれです。実際に使うことを想定してデータベースから取得するイメージをもってコーディングします。

対象とするモデルは、以下の３つでそれぞれ親子関係をサンプルとして使います。

Company
- company_id
- name
- service
Employee
- employee_id
- name
- company_id（上述のCompanyのどれかに所属する）
Task
- task_id
- name
- due_date
- employee_id（上述のEmployeeのどれかに所属する）

CompanyとEmployeeの関係は1:Nで、EmployeeとTaskの関係も1:Nです。

まずはsrc/models/company.py です。ピュアなモデルを表すCompanyクラスと、データベースから取得する処理の代わりを書いているCompanyRepositoryクラスと、GraphQLのスキーマを書いているCompanySchemaクラスがあります。

Companyはcompany_idを通して複数のEmployeeを持っているのでresolve_employeesで、自身のcompany_idに関連するEmployeeを取得するようにします。

from dataclasses import dataclass
from graphene import ObjectType, ID, String, List
from typing import Dict

from .employee import EmployeeSchema, EmployeeRepository


@dataclass
class Company:
   company_id: int
   name: str
   service: str


class CompanyRepository:
   companies: Dict = {}

   @classmethod
   def find(cls, company_id: int):
       return cls.companies[company_id]

   @classmethod
   def create(cls):
       cls.companies = {
           company_id: Company(company_id, f'company-{company_id}', 'edu-tech')
           for company_id in range(1, 10)
       }


class CompanySchema(ObjectType):
   company_id = ID()
   name = String()
   service = String()
   employees = List(EmployeeSchema)

   def resolve_employees(self, info):
       return EmployeeRepository.find_by_company(self.company_id)

次はsrc/models/employee.pyです。構成はCompanyと同じため説明は割愛します。

from dataclasses import dataclass
from graphene import ObjectType, ID, Int, String, List
from typing import Dict

from .task import TaskSchema, TaskRepository


@dataclass
class Employee:
   employee_id: int
   name: str
   company_id: int


class EmployeeRepository:
   employees: Dict = {}

   @classmethod
   def find_by_company(cls, company_id: int):
       return [employee for employee in cls.employees.values()
               if employee.company_id == company_id]
 
  @classmethod
   def create(cls):
       cls.employees = {
           employee_id: Employee(employee_id, f'employee-{employee_id}', (employee_id % 10) + 1)
           for employee_id in range(1, 30)
       }


class EmployeeSchema(ObjectType):
   employee_id = ID()
   name = String()
   company_id = Int()
   tasks = List(TaskSchema)

   def resolve_tasks(self, info):
       return TaskRepository.find_by_employee(self.employee_id)

次はsrc/models/task.pyです。

from dataclasses import dataclass
from datetime import datetime
from graphene import ObjectType, ID, Int, String
from graphene.types.datetime import DateTime
from typing import Dict


@dataclass
class Task:
   task_id: int
   name: str
   due_date: datetime
   employee_id: int


class TaskRepository:
   tasks: Dict = {}

   @classmethod
   def find_by_employee(cls, employee_id: int):
       return [task for task in cls.tasks.values()
               if task.employee_id == employee_id]

   @classmethod
   def create(cls):
       cls.tasks = {
           task_id: Task(task_id, f'task-{task_id}', (task_id % 30) + 1, datetime.now())
           for task_id in range(1, 100)
       }


class TaskSchema(ObjectType):
   task_id = ID()
   name = String()
   due_date = DateTime()
   employee_id = Int()

最後にserver.pyです。

サーバー起動時に初期データを登録するため@api.on_event('startup')でデータを作成しています。

import responder

from graphene import ObjectType, Schema, Field, Int

from src.models.company import CompanySchema, CompanyRepository
from src.models.employee import EmployeeRepository
from src.models.task import TaskRepository


api = responder.API()

@api.on_event('startup')
def init():
   CompanyRepository.create()
   EmployeeRepository.create()
   TaskRepository.create()

class CompanyQuery(ObjectType):
   company = Field(CompanySchema, company_id=Int())

   def resolve_company(self, info, company_id: int):
       return CompanyRepository.find(company_id)

schema = Schema(query=CompanyQuery)
view = responder.ext.GraphQLView(api=api, schema=schema)

api.add_route("/graph", view)

@api.route("/")
async def index(req, resp):
   resp.media = {"message": 'ok'}

@api.route("/company/")
async def company(req, resp):
   data = await req.media()
   query_string = data['query']
   result = schema.execute(query_string)
   resp.media = {'message': 'ok', 'result': result.to_dict()}

/company/にqueryを送ると、schema.executeでqueryが解釈され、データを取得することができます。

たとえばcurlなら以下のようなリクエストを送ると、同じく下記のようなレスポンスが得られます（見やすいようにjqで加工してます）

curl http://localhost:8000/company/ -H 'Content-Type: application/json' -d '{ "query": "{ company(companyId: 3) { companyId name employees { name tasks { name dueDate } } } }" }'
{
 "message": "ok",
 "result": {
   "data": {
     "company": {
       "companyId": "3",
       "name": "company-3",
       "employees": [
         {
           "name": "employee-2",
           "tasks": [
             {
               "name": "task-1",
               "dueDate": "2019-08-14T00:10:53.678245"
             },
             {
               "name": "task-31",
               "dueDate": "2019-08-14T00:10:53.679039"
             },
             {
               "name": "task-61",
               "dueDate": "2019-08-14T00:10:53.679513"
             },
             {
               "name": "task-91",
               "dueDate": "2019-08-14T00:10:53.679956"
             }
           ]
         },
         {
           "name": "employee-12",
           "tasks": [
             {
               "name": "task-11",
               "dueDate": "2019-08-14T00:10:53.678561"
             },
             {
               "name": "task-41",
               "dueDate": "2019-08-14T00:10:53.679068"
             },
             {
               "name": "task-71",
               "dueDate": "2019-08-14T00:10:53.679728"
             }
           ]
         },
         {
           "name": "employee-22",
           "tasks": [
             {
               "name": "task-21",
               "dueDate": "2019-08-14T00:10:53.679008"
             },
             {
               "name": "task-51",
               "dueDate": "2019-08-14T00:10:53.679218"
             },
             {
               "name": "task-81",
               "dueDate": "2019-08-14T00:10:53.679843"
             }
           ]
         }
       ]
     }
   }
 }
}

まとめ

というわけで、GraphQLについて今回は試してみました。データが多いときはRelayを使ったほうがよいとか、更新するときはMutationを使うとか、なかなかGraphQLも奥が深そうです。RESTful APIのエンドポイントを1つのGraphQLのエンドポイントにしたら結構管理が大変そうな印象もあるけど、実際はどうなんでしょうね。プロダクション環境で利用している人の話を聞いてみたくなりました。

ちなみに試してはないですが、DjangoやSQLAlchemyでは定義したORMのモデルを渡すことで簡単にGraphQL化できるみたいなので、実際はモデルとスキーマで二重で定義するみたいなことしなくて良さそうです。（そもそもメタ情報としてdataclassを渡す方法もあるかも？）

ではでは、今回はこのへんで〜

サポートありがとうございます。頂いたご支援は美味しいものを食べに行きます。