LoginSignup
32

More than 3 years have passed since last update.

PythonでGraphQLチュートリアル+α

Last updated at Posted at 2019-09-17
1 / 16

はじめに

GraphQLのPython用フレームワークgrapheneのチュートリアルとflask-graphene-sqlalchemyがとてもよかったので自分なりにの構成と環境構築をまとめました。

grapheneではなくFlaskとSQLAlchemyを使用を前提としたライブラリFlask-Graphene-SQLAlchemyのチュートリアルを参考に行なって行きます。

そんなにGraphQL詳しくないのでとりあえず写経から初めて、ディレクトリの構成とかチュートリアルにはないが気になるところを追記しています。

作ったもの: https://github.com/thimi0412/Flask-Graphene-SQLAlchemy


環境構築

仮想環境構築にはPipenvを使用

$ pipenv install --python 3.7
$ pipenv install sqlalchemy \
                 graphene-sqlalchemy \
                 flask \
                 flask-graphql

データベースはsqliteを使用。サクッとチュートリアルをやってみたかったんでDocker使ってなくてすまん。


ディレクトリ構成

├── Pipfile  #みんなご存知Pipfile
├── Pipfile.lock
├── README.md
├── app.py   #サーバ
├── database  #データベース情報
│   ├── base.py
│   ├── model_department.py
│   ├── model_employee.py
│   └── model_role.py
├── database.sqlite3
├── schema  #GraphQlの中核を担う部分
│   ├── schema.py
│   ├── schema_department.py
│   ├── schema_employee.py
│   └── schema_role.py
└── setup.py  #初期データ登録用

チュートリアルではdatabese, schemaはそれぞれ1ファイルに記述されていたのでモデルごとに分割。


データベース作成

model_employee.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker

engine = create_engine('sqlite:///database.sqlite3', convert_unicode=True)

Base = declarative_base()
Base.metadata.bind = engine

db_session = scoped_session(sessionmaker(bind=engine, expire_on_commit=False))
Base.query = db_session.query_property()

データベースの情報を記述します。


テーブル定義

model_employee.py
from .base import Base
from .model_department import ModelDepartment
from .model_role import ModelRole
from sqlalchemy import (
    Column, DateTime, ForeignKey, Integer, String, func
)
from sqlalchemy.orm import backref, relationship


class ModelEmployee(Base):
    __tablename__ = 'employee'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    hired_on = Column(DateTime, default=func.now())
    department_id = Column(Integer, ForeignKey('department.id'))
    role_id = Column(Integer, ForeignKey('roles.id'))
    department = relationship(
        ModelDepartment,
        backref=backref('employees', uselist=True, cascade='delete,all'))
    role = relationship(
        ModelRole,
        backref=backref('roles', uselist=True, cascade='delete,all'))
model_department.py
from .base import Base
from sqlalchemy import Column, Integer, String


class ModelDepartment(Base):
    __tablename__ = 'department'
    id = Column(Integer, primary_key=True)
    name = Column(String
model_role.py
from .base import Base
from sqlalchemy import Column, Integer, String


class ModelRole(Base):
    __tablename__ = 'roles'
    id = Column(Integer, primary_key=True)
    name = Column(String)

チュートリアルではEmployee(従業員), Department(部門), Role(役職)のテーブルを作成し、リレーションを張っています。
ここは、SQLAlchemyのテーブルの作成とかわりありません。


テストデータを入れる

setup.py
from database.model_department import ModelDepartment
from database.model_role import ModelRole
from database.model_employee import ModelEmployee
from database import base
import logging
import sys


log = logging.getLogger(__name__)
logging.basicConfig(
    stream=sys.stdout,
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)


if __name__ == "__main__":
    log.info('Create database')
    base.Base.metadata.create_all(base.engine)

    log.info('Insert data')
    engineering = ModelDepartment(name='Engineering')
    base.db_session.add(engineering)
    hr = ModelDepartment(name='Human Resources')
    base.db_session.add(hr)

    manager = ModelRole(name='manager')
    base.db_session.add(manager)
    engineer = ModelRole(name='engineer')
    base.db_session.add(engineer)

    peter = ModelEmployee(name='Peter', department=engineering, role=engineer)
    base.db_session.add(peter)
    roy = ModelEmployee(name='Roy', department=engineering, role=engineer)
    base.db_session.add(roy)
    tracy = ModelEmployee(name='Tracy', department=hr, role=manager)
    base.db_session.add(tracy)
    base.db_session.commit()

チュートリアルでは対話モードでデータを入れていましたがGitHubのリポジトリではスクリプトを作成して、データを入れていたので参考にして作成。とりあえずこれを実行すればOK。
テーブルは以下のように作成されます。

sqlite> select * from employee;
1|Peter|2019-09-09 03:08:31|1|2
2|Roy|2019-09-09 03:08:31|1|2
3|Tracy|2019-09-09 03:08:31|2|1

sqlite> select * from department;
1|Engineering
2|Human Resources

sqlite> select * from roles;
1|manager
2|engineer

スキーマを作成する

schema_employee.py
from database.model_employee import ModelEmployee
import graphene
from graphene import relay
from graphene_sqlalchemy import SQLAlchemyConnectionField, SQLAlchemyObjectType


class EmployeeAttribute:
    name = graphene.String(description='Name of Role')
    hired_on = graphene.DateTime(description='Hired_on date')
    department_id = graphene.Int(description='ID of Department')
    role_id = graphene.Int(description='ID of Role')


class Employee(SQLAlchemyObjectType, EmployeeAttribute):
    class Meta:
        model = ModelEmployee
        interfaces = (relay.Node,)


Model = ModelEmployee

schema_department.py
from database.model_department import ModelDepartment
import graphene
from graphene import relay
from graphene_sqlalchemy import SQLAlchemyConnectionField, SQLAlchemyObjectType


class DepertmentAttribute:
    name = graphene.String(description='Name of Department')


class Department(SQLAlchemyObjectType, DepertmentAttribute):
    class Meta:
        model = ModelDepartment
        interfaces = (relay.Node,)


Model = ModelDepartment
schema_role.py
from database.model_role import ModelRole
import graphene
from graphene import relay
from graphene_sqlalchemy import SQLAlchemyConnectionField, SQLAlchemyObjectType


class RoleAttribute:
    name = graphene.String(description='Name of Role')


class Role(SQLAlchemyObjectType, RoleAttribute):
    class Meta:
        model = ModelRole
        interfaces = (relay.Node,)


Model = ModelRole

Queryを作成

schema.py
from graphene_sqlalchemy import SQLAlchemyConnectionField
from graphene import relay
import graphene
from . import schema_department
from . import schema_role
from . import schema_employee


class Query(graphene.ObjectType):
    node = relay.Node.Field()
    employee_list = SQLAlchemyConnectionField(schema_employee.Employee)
    department_list = SQLAlchemyConnectionField(schema_department.Department)
    role_list = SQLAlchemyConnectionField(schema_role.Role)
schema = graphene.Schema(query=Query)

REST APIで言うところのGET /hogehogeをここで作成指定ます。Queryクラス内でemployee_list, department_list, role_listをの関数を定義指定ます。

ちなみに関数をスネークケースで定義していますがGraphQLではキャメルケースとして解釈されます。


URLを設定

app.py
from database.base import db_session
from flask import Flask
from flask_graphql import GraphQLView
from schema.schema import schema

app = Flask(__name__)
app.add_url_rule(
    '/graphql',
    view_func=GraphQLView.as_view('graphql', schema=schema, graphiql=True)
)


@app.teardown_appcontext
def shutdown_session(exception=None):
    db_session.remove()


if __name__ == "__main__":
    app.run(threaded=True, debug=True)

GraphQLにアクセスするURLは1つだけなので/graphql配下にschemaを設定します。


Queryを投げてみる

サーバを起動

pipenv run python app.py

http://localhost:5000/graphql にアクセスするして画面左の入力画面に以下のクエリを記述して右上のRUNボタンを押すとクエリが実行されます。
ここがすごい! コピペしてもいいですが補完がすごいので比較的楽に入力できる

query{
  employeeList{
    edges{
      node{
        name
        hiredOn
        department{
          name
        },
        role{
          name
        }
      }
    }
  }
}

スクリーンショット 2019-09-16 21.28.23.png

とりあえずチュートリアルの内容がここまでとなっていました。なので以降は個人的に必要そうな内容です。


[追加]従業員名で検索したい

従業員全ての情報は取得することができたので名前で検索して、情報を取得します。 個人的にここがハマって辛かった。
schema.pyを編集します。

schema.py
class Query(graphene.ObjectType):
    node = relay.Node.Field()
    # 追加
    employee = graphene.Field(lambda: schema_employee.Employee,
                              name=graphene.String())
    employee_list = SQLAlchemyConnectionField(schema_employee.Employee)
    department_list = SQLAlchemyConnectionField(schema_department.Department)
    role_list = SQLAlchemyConnectionField(schema_role.Role)

    # 追加
    def resolve_employee(self, info, name):
        query = schema_employee.Employee.get_query(info)
        result = query.filter(schema_employee.Model.name == name).first()
        return result


schema = graphene.Schema(query=Query)

employeeと言う関数を追加して、employeeが呼ばれた時実際に処理を行う関数resolve_employeeを作成します。これはresolve_<関数名>と解釈されるのでclassで追加した関数名と同じである必要があります。

ここで注意したいのは、resolve_employeeで実際にSQLAlchemyで名前で検索を行なっているところです。
最初はschema_employee.Employeeを使用してfilterで検索をしようとしていましたが、これはGraphQLを解釈するたのクラスのなので、SQLAlchemyで作成したクラス(databese/model_employee.py)のクラスを使用する必要があります。

なのでschema/schema_employee.pyの最後の行にdatabese/model_employee.pyクラスをModelとして定義しています。

Model = ModelEmployee

早速検索

query{
  employee(name: "Roy"){
    name
    department{
      name
    }
    role{
      name
    }
  }
}

スクリーンショット 2019-09-16 22.52.32.png


[追加]データを追加したい

データを追加する場合はMutateを使用します。REST APIPOST/PUTに相当。 shema/schema.pyを編集します。

schema.py
from graphene_sqlalchemy import SQLAlchemyConnectionField
from graphene import relay
import graphene
from . import schema_department
from . import schema_role
from . import schema_employee
from database import base # 追加 db_sessionを使用するため


# 追加
class InsertEmployee(graphene.Mutation):
    # 貰うパラメータの設定
    class Arguments:
        name = graphene.String(required=True)
        department = graphene.String(required=True)
        role = graphene.String(required=True)

    employee = graphene.Field(lambda: schema_employee.Employee)

    # テーブルにインサートする処理
    def mutate(self, info, name, department, role):
        department = schema_department.Model(name=department)
        base.db_session.add(department)
        role = schema_role.Model(name=role)
        base.db_session.add(role)

        employee = schema_employee.Model(
            name=name, department=department, role=role)
        base.db_session.add(employee)

        base.db_session.commit()
        return InsertEmployee(employee=employee)


class Query(graphene.ObjectType):
    node = relay.Node.Field()
    employee = graphene.Field(lambda: schema_employee.Employee,
                              name=graphene.String())
    employee_list = SQLAlchemyConnectionField(schema_employee.Employee)
    department_list = SQLAlchemyConnectionField(schema_department.Department)
    role_list = SQLAlchemyConnectionField(schema_role.Role)

    def resolve_employee(self, info, name):
        query = schema_employee.Employee.get_query(info)
        result = query.filter(schema_employee.Model.name == name).first()
        return result


# 追加
class Mutation(graphene.ObjectType):
    insert_employee = InsertEmployee.Field()


# mutationを追加
schema = graphene.Schema(query=Query, mutation=Mutation)

InsertEmployeeのクラスを作成し、クラス内にパラメータを受け取るクラスArguments、実際に処理を行う関数mutateを作成します。

ここも先ほどの名前で検索と同様にSQLAlchemyで作成したクラスを使用します。

そしてMutationクラスを作成し、schemaに追加します。


データを追加

以下のクエリでデータを追加

mutation{
  insertEmployee(name: "Kosuke", department: "R&D", role: "chief doctor"){
    employee{
      name
      department{
        name
      }
      role{
        name
      }
    }
  }
}

スクリーンショット 2019-09-17 0.06.23.png

確認としてemployeeListで最後の1件を取得します。
スクリーンショット 2019-09-17 0.09.41.png
ちゃんとインサートできています。


GraphiQLがすごい

スクリーンショット 2019-09-17 10.32.54.png
Historyのボタンを押す今まで実行したクエリの履歴をみることができます。(地味にこれが便利)
お気に入り機能もあるのでよく使うクエリは登録することができます。

ちなみにどこに履歴情報はブラウザのローカルストレージに格納されているようです。
スクリーンショット 2019-09-17 10.37.53.png

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
32