FastAPI・Graphene・SQLAlchemyでGraphQLサーバー（Relay対応）を構築する

VISITS Technologies Advent Calendar 2020 15日目は@woods0918が担当します。

私がプロダクトマネージャーを務めているideagramというサービスのバックエンドでは、REST APIからGraphQLへの移行が進んでいます。

[初めてのGraphql](https://www.oreilly.co.jp/books/9784873118932/)や[GraphQL Playground](https://github.com/graphql/graphql-playground)を通してGraphQLの知識とクライアント側の操作は経験したのですが、GraphQLサーバーを実装した経験がなかったので、バックエンドエンジニアとの会話をスムーズにするために、挑戦してみようと思います。

今回は書き慣れているPythonで実装しますが、ideagramで採用している言語はRuby（WAF : Ruby on Rails）です。

コードは[こちら](https://github.com/woods0918/graphql_server_sample)

## Pythonとライブラリのバージョン
```pyproject.toml
python = "^3.8"
fastapi = "^0.62.0"
uvicorn = "^0.13.0"
SQLAlchemy = "^1.3.20"
graphene = "^2.1.8"
python-dotenv = "^0.15.0"
mysqlclient = "^2.0.2"
graphene-sqlalchemy = "^2.3.0"
```

## ディレクトリ構成
```
# 説明に必要なファイルのみ抜粋
|--.env # データベース接続のためのパラメータを定義
|--app
  |--__init__.py
  |--cruds # データベースへ投げるクエリを定義
    |--__init__.py
    |--seeds.py
    |--todos.py
    |--users.py
  |--database.py # データベースとの接続を定義
  |--main.py # FastAPIのエンドポイント
  |--models # SQLAlchemyで利用するクラスを定義
    |--__init__.py
    |--todos.py
    |--users.py
  |--schema # Grapheneで利用するQuery, Mutation, Schemaなどを定義
    |--__init__.py
    |--schema.py
    |--todos.py
    |--users.py
  |--setting.py
```

## database.py
```database.py
import os
from contextlib import contextmanager
from sqlalchemy import create_engine, inspect
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker
from dotenv import load_dotenv

load_dotenv()

DB_USERNAME = os.getenv("DB_USERNAME")
DB_PASSWORD = os.getenv("DB_PASSWORD")
DB_HOSTNAME = os.getenv("DB_HOSTNAME")
DB_PORT = os.getenv("DB_PORT")
DB_NAME = os.getenv("DB_NAME")
DB_CHARSET = os.getenv("DB_CHARSET")

ENGINE = create_engine(
    f"mysql://{DB_USERNAME}:{DB_PASSWORD}@{DB_HOSTNAME}:{DB_PORT}/{DB_NAME}?charset={DB_CHARSET}"
)
SESSION = scoped_session(
    sessionmaker(autocommit=False, autoflush=False, bind=ENGINE, expire_on_commit=False)
)

class SelfBase(object):
    def to_dict(self):
        model = {}
        for column in self.__table__.columns:
            model[column.name] = str(getattr(self, column.name))
        return model

BASE = declarative_base(cls=SelfBase)
BASE.query = SESSION.query_property()

@contextmanager
def session_scope():
    session = SESSION()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()
```

database.pyではデータベースとの接続周りを定義します。
DBのパラメーターは、dotenvのload_dotenvを利用して`.env`より読み取ります。

SESSIONは`scoped_session `と`sessionmaker `を利用して生成します。
`scoped_session `を利用すると、何度もSession()を実行したとしても同じセッションが返却されます。

SelfBaseクラスは、（後ほど作成する）SQLAlchemyで利用するクラスを辞書型に変換するための基底クラスです。
sqlalchemyの基底クラスは、SelfBaseクラスを継承して生成します。

`BASE.query = SESSION.query_property()`をやっておくと、
`session.query(HogeClass).filter(HogeClass.col1 == "hoge").all()`
を
`HogeClass.query.filter(HogeClass.col1 == "hoge").all()`
のように書けるので、地味に便利です。

最後に、`session_scope`は[SQLAlchemyでのsessionの扱い方](https://qiita.com/ariku/items/14f49fe24395f1dd9412)の記事を参考にさせていただきました。

```
with session_scope() as session:
    # 何かしらのDB処理
```
という風に書いてあげると、いちいちcommitやcloseを書かなくて良くなるので便利だなと思いました。

## models
modelsでは、SQLAlchemyで利用するクラスを定義します。Userクラスを例にします。

```users.py
from sqlalchemy import Column
from sqlalchemy.dialects.mysql import INTEGER as Integer
from sqlalchemy.dialects.mysql import TEXT as Text
from app.database import BASE

class User(BASE):
    __tablename__ = "users"

    id = Column(Integer(unsigned=True), primary_key=True, unique=True, autoincrement=True)
    name = Column(Text)

    def __init__(self, name: str):
        self.name = name
```
先ほど、database.pyで生成したsqlalchemyの基底クラスを継承して生成します。

`__tablename__`にDBの実際のテーブル名を定義します。

`id`と`name`はテーブルのカラムです。sqlalchemyからデータ型をインポートして定義していきます。
MySQL独自の機能を利用する場合は、`sqlalchemy.dialects.mysql`よりインポートしてください。

最後に、`__init__`で初期化処理を定義します。

## cruds
### seeds.py
seeds.pyでは、テーブルのCREATEとシードデータのINSERTを行います。

```seeds.py
import sys
import pathlib
from datetime import datetime

current_dir = pathlib.Path(__file__).resolve().parent
sys.path.append( str(current_dir) + '/../../' )

from app.database import BASE, ENGINE, session_scope
from app.models.todos import Todo
from app.models.users import User

def generate_seed_data():
    BASE.metadata.create_all(ENGINE)
    users = [["太郎"], ["次郎"], ["花子"]]
    todos = [
        [1, "title1", "description1", datetime.now()],
        [1, "title2", "description2", datetime.now()],
        [2, "title3", "description3", datetime.now()],
        [2, "title4", "description4", datetime.now()],
        [3, "title5", "description5", datetime.now()],
        [3, "title6", "description6", datetime.now()]
    ]

    with session_scope() as session:
        for user in users:
            session.add(User(user[0]))
        for todo in todos:
            session.add(Todo(
                user_id = todo[0],
                title = todo[1],
                description = todo[2],
                deadline = todo[3]
            ))

if __name__ == "__main__":
    generate_seed_data()
```
まず、`BASE.metadata.create_all(ENGINE)`でテーブルを生成します。
既存のテーブルから変更がない場合はスキップして作ってくれたり、「よしなに」CREATEを実行してくれるので、非常に便利です。

その後、Seedデータを生成し、`with session_scope() as session:`以降でINSERTを行います。

### users.py
（後ほど）、QueryやMutationを行う際に利用するクエリを定義します。

```users.py
from typing import Dict, Optional
from app.database import session_scope
from app.models.users import User

def insert_user(name: str) -> User:
    user = User(name)
    with session_scope() as session:
        session.add(user)
        session.flush()
        return user

def fetch_user(id: Optional[int]=None, name: Optional[str]=None):
    with session_scope() as session:
        query = session.query(User)
        if id:
            query = query.filter(User.id == id)
        if name:
            query = query.filter(User.name == name)
        return query.all()
```
名前の通りですが、`insert_user `ではINSERTを、`fetch_user`ではSELECTを行います。

## schema
いよいよ、GraphQLの部分に入ります。
### users.py
`users.py`や`todos.py`では、先ほどmodelsで生成したクラスに対応したGraphQL用のクラスを生成します。

```users.py
import graphene
from graphene import relay
from graphene_sqlalchemy import SQLAlchemyObjectType

from app.models.users import User as UserModel
from app.cruds.users import insert_user


#-------------------------
# Query
#-------------------------
class User(SQLAlchemyObjectType):
    class Meta:
        model = UserModel
        interface = (relay.Node, )

class UserConnections(relay.Connection):
    class Meta:
        node = User

#-------------------------
# Mutation
#-------------------------
class InsertUser(relay.ClientIDMutation):
    class Input:
        name = graphene.String(required=True)
    
    user = graphene.Field(User)

    @classmethod
    def mutate_and_get_payload(cls, root, info, name):
        user = insert_user(name)
        return InsertUser(user)
```
#### Query
`class User`は`SQLAlchemyObjectType`を継承して生成します。
さらに、`model = UserModel`でSQLAlchemyのクラスを紐づけることで、GraphQLとSQLAlchemyの間を上手く繋いでくれます。
詳細は、[graphene-sqlalchemyのリポジトリ](https://github.com/graphql-python/graphene-sqlalchemy)を参照してください。

さらに、`interface = (relay.Node, )`や`UserConnections `を定義することで、簡単にRelay対応を行うことができます。
詳しくは、[grapheneのrelayに関するドキュメント](https://docs.graphene-python.org/en/latest/relay/)を参照してください。

#### Mutation
`class InsertUser `は`relay.ClientIDMutation`を継承して生成します。
`Input`で`InsertUser`を実行する際の引数を定義します。
`mutate_and_get_payload `では、実際にDBにアクセスする処理を定義します。（先ほど、crudsのusersで定義した`insert_user`関数を利用します。）

### schema.py
`schema.py`では、QueryやMutationのRootとなる部分を定義します。

```schema.py
import graphene
from graphene import relay
from graphene_sqlalchemy import SQLAlchemyObjectType, SQLAlchemyConnectionField

from app.cruds.users import fetch_user
from app.cruds.todos import fetch_todo
from app.schema.users import User, UserConnections, InsertUser
from app.schema.todos import Todo, TodoConnections, InsertTodo

#-------------------------
# Query
#-------------------------
class Query(graphene.ObjectType):
    node = relay.Node.Field()

    user = graphene.Field(
        lambda: graphene.List(User),
        id=graphene.Int(required=False),
        name=graphene.String(required=False)
    )
    todo = graphene.Field(
        lambda: graphene.List(Todo),
        id=graphene.Int(required=False),
        user_id=graphene.Int(required=False),
        title=graphene.String(required=False),
        description=graphene.String(required=False),
        deadline=graphene.DateTime(required=False)
    )
    all_users = SQLAlchemyConnectionField(UserConnections)
    all_todos = SQLAlchemyConnectionField(TodoConnections, sort=None)

    def resolve_user(self, info, id=None, name=None):
        return fetch_user(id, name)
    
    def resolve_todo(self, info, id=None, user_id=None, title=None, description=None, deadline=None):
        return fetch_todo(id, user_id, title, description, deadline)

#-------------------------
# Mutation
#-------------------------
class Mutation(graphene.ObjectType):
    insert_user = InsertUser.Field()
    insert_todo = InsertTodo.Field()

#-------------------------
# Schema
#-------------------------
schema = graphene.Schema(query = Query, mutation=Mutation)
```
#### Query
`node = relay.Node.Field()`
`all_users = SQLAlchemyConnectionField(UserConnections)`
`all_todos = SQLAlchemyConnectionField(TodoConnections, sort=None)`
は、relayの規格に沿ったクエリを投げられるようにするためのものです。

`user`と`resolve_user `、`todo`と`resolve_todo `は対応するもので、マニュアルで定義したSELECT処理をGraphQLとして扱えるようにするためのものです。

#### Mutation
Mutationクラスに`users.py`や`todos.py`で生成したMutationクラスを紐づけていきます。

#### Schema
RootのQueryクラスとMutationクラスをSchemaに紐付けます。

## main.py
最後に、FastAPIのエンドポイントを生成します。

```main.py
import sys, pathlib, uvicorn, graphene

current_dir = pathlib.Path(__file__).resolve().parent
sys.path.append( str(current_dir) + '/../' )

from graphql.execution.executors.asyncio import AsyncioExecutor
from fastapi import FastAPI
from starlette.graphql import GraphQLApp
from app.schema.schema import schema

app = FastAPI()
app.add_route("/", GraphQLApp(schema=schema))

if __name__ == "__main__":
    uvicorn.run("main:app", port=5000, reload=True, access_log=False)
```
`FastAPI()`でappインスタンスを生成し、`/`でGraphQLにアクセスできるようにroute設定を行います。

## 実行
以下のコマンドで起動できます。

```shell

python ./app/main.py
```

## 結果確認
今回は、[GraphQL Playground](https://github.com/graphql/graphql-playground)を使って結果を確認していきます。

### Schema
GraphQL Playground上では以下のように表示されます。
![スクリーンショット 2020-12-14 22.36.03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/270069/2222ad15-b6a4-b335-6cf8-a505df5c1e52.png)


Schema全体は以下のようになります。

```graphql
scalar DateTime

input InsertTodoInput {
  userId: Int!
  title: String!
  description: String!
  deadline: DateTime!
  clientMutationId: String
}

type InsertTodoPayload {
  todo: Todo
  clientMutationId: String
}

input InsertUserInput {
  name: String!
  clientMutationId: String
}

type InsertUserPayload {
  user: User
  clientMutationId: String
}

type Mutation {
  insertUser(input: InsertUserInput!): InsertUserPayload
  insertTodo(input: InsertTodoInput!): InsertTodoPayload
}

# An object with an ID
interface Node {
  # The ID of the object.
  id: ID!
}

# The Relay compliant `PageInfo` type, containing data necessary to paginate this connection.
type PageInfo {
  # When paginating forwards, are there more items?
  hasNextPage: Boolean!
  # When paginating backwards, are there more items?
  hasPreviousPage: Boolean!
  # When paginating backwards, the cursor to continue.
  startCursor: String
  # When paginating forwards, the cursor to continue.
  endCursor: String
}

type Query {
  # The ID of the object
  node(id: ID!): Node
  user(id: Int, name: String): [User]
  # <graphene.types.scalars.String object at 0x111cd6ac0>
  todo(id: Int, userId: Int, title: String, deadline: DateTime): [Todo]
  allUsers(
    sort: [UserSortEnum]
    before: String
    after: String
    first: Int
    last: Int
  ): UserConnectionsConnection
  allTodos(
    before: String
    after: String
    first: Int
    last: Int
  ): TodoConnectionsConnection
}

type Todo {
  id: ID!
  userId: Int
  title: String
  description: String
  finished: Boolean
  deadline: DateTime
  users: User
}

type TodoConnectionsConnection {
  # Pagination data for this connection.
  pageInfo: PageInfo!
  # Contains the nodes in this connection.
  edges: [TodoConnectionsEdge]!
}

# A Relay edge containing a `TodoConnections` and its cursor.
type TodoConnectionsEdge {
  # The item at the end of the edge
  node: Todo
  # A cursor for use in pagination
  cursor: String!
}

type User {
  id: ID!
  name: String
}

type UserConnectionsConnection {
  # Pagination data for this connection.
  pageInfo: PageInfo!
  # Contains the nodes in this connection.
  edges: [UserConnectionsEdge]!
}

# A Relay edge containing a `UserConnections` and its cursor.
type UserConnectionsEdge {
  # The item at the end of the edge
  node: User
  # A cursor for use in pagination
  cursor: String!
}

# An enumeration.
enum UserSortEnum {
  ID_ASC
  ID_DESC
  NAME_ASC
  NAME_DESC
}
```
PageInfoやinterface NodeなどはRelayに関わるところですね。

他にもクエリの結果を見ていきます。

### allUsers
![スクリーンショット 2020-12-14 22.42.41.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/270069/66aa7704-a7c0-f71c-9f38-bf1ebe2aedaa.png)

### allTodos
![スクリーンショット 2020-12-14 22.44.18.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/270069/378d4515-e34f-7f5b-9c56-f99126cc817e.png)

### user(resolve_user)
![スクリーンショット 2020-12-14 22.45.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/270069/7ce0dcc3-5ee0-08bd-9883-9b33bbb964e9.png)

### todo(resolve_todo)
![スクリーンショット 2020-12-14 22.46.20.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/270069/300b307d-0c69-94af-72ea-eb52617d7a91.png)

### insertUser
![スクリーンショット 2020-12-14 22.49.53.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/270069/d5f8d337-3fdf-ca3d-e399-d6a733854c66.png)

### insertTodo
![スクリーンショット 2020-12-14 22.53.19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/270069/ef8e3c69-56ff-8ea6-5687-b12ff1dd4a02.png)


## おわりに
はじめて、GrpahQLのサーバーサイドを構築しましたが、REST APIより難しいと感じるのは慣れていないからでしょうかね。
今度、同僚のエンジニアにコツを教えてもらおうと思います。

久しぶりにコードを書いたので、あまり綺麗に書けていなかったり理解が誤っている箇所があるかもしれませんが、その際はこそっとコメント欄で教えてもらえると幸いです。

明日は、ideagramのGraphQL化を1から進めた[@ham0215](https://qiita.com/ham0215)による記事です、お楽しみに！


## 参考
### 公式
[SQLAlchemy 1.3 Documentation](https://docs.sqlalchemy.org/en/13/index.html)
[Graphene Documentaion](https://graphene-python.org/)
[FastAPI Documentaion](https://fastapi.tiangolo.com/ja/)
[graphene-sqlalchemyリポジトリ](https://github.com/graphql-python/graphene-sqlalchemy)

### 記事
[SQLAlchemyでのsessionの扱い方](https://qiita.com/ariku/items/14f49fe24395f1dd9412)
[PythonでGraphQLチュートリアル+α](https://qiita.com/thimi0412/items/a4234145c1a98459fe6d)
[FastAPI ディレクトリ設計](https://note.com/yusugomori/n/n9f2c0422dfcd)
[sqlalchemyのテーブル定義tips](https://qiita.com/petitviolet/items/e03c67794c4e335b6706)