はじめに
最近アクアリウムを始めました、菅野です。
プログラムと異なり、生体を扱う都合上の想定外を楽しみながら試行錯誤しております。
さて、皆さんはAPIサーバを構築する際に、どのAPI形式を用いていますか？
まだまだREST形式で実装することが多いかとは思いますが、
GraphQLを用いることも増えてきているのではないでしょうか？
今回は、そんなGraphQLをFastAPIと各種ライブラリを用いて簡単に実装する方法を紹介していこうと思います。
GraphQLとは
GraphQLは、Meta社（旧Facebook社）によって開発・公開されたAPI仕様です。クエリ形式で、処理やパラメータの内容を指定します。
RESTとの比較としては、
- クライアント側で取得したい情報をクエリとして渡すことができるため、 利用しないデータを無駄に受け取らなくて済む。
- 一つのエンドポイントに対し複数リソースのクエリを一度にリクエストできるので、APIコール数を削減できる
等の利点があります。
GraphQL | A query language for your API
構成
PythonのAPI作成フレームワーク、FastAPIと、GraphQLライブラリStrawberryを組み合わせることで、簡単にGraphqlAPIを実現することができます。
今回は、上記ライブラリに加え、ORMライブラリSQLAlchemyを用いてSQLite3に接続するTODO管理アプリのバックエンドサーバを構築します。
利用ライブラリ
利用するPythonおよび主なライブラリとそのバージョンは以下になります。
| ライブラリ |
バージョン |
| Python |
3.9.6 |
| FastAPI |
0.85.0 |
| Strawberry |
0.131.1 |
| SQLAlchemy |
1.4.41 |
実装
ディレクトリ構成は以下のようになっています。
クリーンアーキテクチャの構成に従いディレクトリを分けております。
/
│ poetry.lock
│ pyproject.toml
├─db
└─src
│ app.py
│ contexts.py
│ database.py
│ router.py
│ resolvers.py
│ __init__.py
│
├─domain
│ ├─model
│ │ task.py
│ │
│ └─service
│ task_service.py
│
├─infra
│ └─repository
│ models.py
│ task_repository.py
│
└─web
└─task
inputs.py
types.py
クラス図
それぞれのコードの関係性は以下になっています。
各種コード説明
infra/repositoryディレクトリ
データベースに接続するためのデータモデル定義と、各種DB操作の実装をこちらに記載します。
models.py
DBテーブルで利用するカラム定義を記載します。
一般的に、SQLAlchemyで用いるDBデータモデルクラスはDeclarative Extensionsを用いて、
テーブルとドメインモデルクラスのマッピングを行います。
しかし、上記を利用すると、ドメインモデルクラスがSQLAlchemyに依存する形になってしまうため、
今回は利用せず、ドメインモデルクラスをSQLAlchemyに依存しないように分けて実装します。
Declarative Extensions — SQLAlchemy 1.4 Documentation
from domain.model.task import Task ,Status
from sqlalchemy import (Column, DateTime, Enum,
String, Table)
from sqlalchemy.orm import registry
from sqlalchemy.sql import func
from sqlalchemy_utils import UUIDType, ChoiceType
import uuid
import enum
mapper_registry = registry()
task = Table(
'task',
mapper_registry.metadata,
Column('id', UUIDType(binary=False), primary_key=True, default=uuid.uuid4),
Column('description', String(200)),
Column('title', String(200)),
Column('status', Enum(Status)),
Column('updated_at', DateTime, default=func.now())
)
mapper_registry.map_imperatively(Task, task)
task_repository.py
DBセッション情報を受け取り、CRUD操作を行う処理を記載します。
from sqlalchemy.orm import Session
from domain.model.task import Task
class TaskRepository:
def __init__(self, db: Session):
self.__db = db
def find_by_id(self, id_: str) -> Task:
task = self.__db.query(Task).get(id_)
return task
def find_all(self) -> list[Task]:
tasks = self.__db.query(Task).all()
return tasks
def save(self, task: Task) -> None:
self.__db.add(task)
self.__db.commit()
def delete(self, task: Task) -> None:
self.__db.delete(task)
self.__db.commit()
内部で行うビジネスロジック（今回ではAPI実行時の各種処理）とデータ型の定義を行います。
service/task_service.py
データアクセスをリポジトリクラスに委任することで接続先に依存しない実装になっています。
例えば、DB接続ではなくオンメモリでデータを保持するようになった場合はリポジトリクラスのみの実装を変更すればよくなります。
from datetime import datetime
from typing import Optional
from domain.model.task import Status, Task
from infra.repository.task_repository import TaskRepository
class TaskService:
def __init__(self, repo: TaskRepository) -> None:
self.__repo = repo
@property
def repo(self) -> type[TaskRepository]:
return self.__repo
def find(self, id: str) -> Task:
task = self.repo.find_by_id(id)
return task
def find_all(self) -> list[Task]:
tasks = self.repo.find_all()
return tasks
def create(self, *, title: str, description: Optional[str] = None) -> Task:
task = Task(title=title, description=description)
self.repo.save(task)
return task
def update(self, *, id: str, status: Status) -> Task:
task = self.repo.find_by_id(id)
task.status = status
task.updated_at = datetime.utcnow()
self.repo.save(task)
return task
def delete(self, id: str) -> Task:
task = self.repo.find_by_id(id)
self.repo.delete(task)
return task
model/task.py
DB接続モデルクラスでも説明したように、SQLAlchemyに依存しないドメインモデルクラスとして、
利用するデータのモデルクラスを別途実装します。
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional
class Status(Enum):
TODO = 'todo'
DOING = 'doing'
DONE = 'done'
@dataclass
class Task:
title: str
id: uuid.UUID = field(default_factory=uuid.uuid4)
description: Optional[str] = None
status: Status = Status.TODO
updated_at: datetime = field(default_factory=datetime.utcnow)
DB接続、ドメインロジックの実装が終わったので、
いよいよ本投稿の要旨であるStrawberryを用いたGraphql実装部分の肝である、webクラスを実装していきます。
types.py
GraphQLでは、クエリを構造を定義する必要があります。
Strawberryではstrawberry.typeアノテーションを用いてコードベースでクエリスキーマを定義できます。
Python型とGraphQLスキーマ型との対応は以下のリンクを参照してください。
Schema basics | 🍓 Strawberry GraphQL
from datetime import datetime
from typing import Optional
import strawberry
from domain.model.task import Status, Task
StatusType = strawberry.enum(Status, name='Status')
@strawberry.type(name='Task')
class TaskType:
id: strawberry.ID
title: str
description: Optional[str]
status: StatusType
updated_at: datetime
input.py
GraphQLリクエストの入力も同様にコードベースで定義します。
strawberry.inputアノテーションを付けたクラスが入力スキーマとして利用できるようになります。
from typing import Optional
import strawberry
@strawberry.input
class AddTaskInput:
title: str
description: Optional[str] = None
@strawberry.input
class UpdateTaskInput:
id: strawberry.ID
status: str
FastAPI起動用のコードを記載します。
database.py
DB接続情報と初回起動時のテーブル初期化処理を記載します。
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from infra.repository.models import mapper_registry
SQLALCHEMY_DATABASE_URI = 'sqlite:///../db/tasks.db'
class DatabaseContext:
def initialize(self):
engine = create_engine(
SQLALCHEMY_DATABASE_URI, connect_args={'check_same_thread': False}, echo=True
)
self.SessionLocal = sessionmaker(
autocommit=False, autoflush=False, bind=engine)
Base = mapper_registry.generate_base()
Base.metadata.create_all(bind=engine)
database_context = DatabaseContext()
def get_db():
"""
Get database
Yields:
SessionLocal: Local session for database connection
"""
db = database_context.SessionLocal()
try:
yield db
finally:
db.close()
context.py
コンテキスト情報として、各種サービスとリポジトリクラスの関係性を保持するように定義します。
from fastapi import Depends
from strawberry.fastapi import BaseContext
from database import get_db
from domain.service.task_service import TaskService
from infra.repository.task_repository import TaskRepository
def init_task_repository(db=Depends(get_db)):
return TaskRepository(db)
def init_task_service(task_repository: TaskRepository = Depends(init_task_repository)):
return TaskService(
task_repository
)
class TaskContext(BaseContext):
def __init__(self, task: TaskService):
self.__task: TaskService = task
def get_task(self):
return self.__task
class TaskServicesContext(BaseContext):
def __init__(self, task: TaskService):
self.__task: TaskService = task
def get_task(self):
return self.__task
async def get_context(
task_service: TaskService = Depends(init_task_service)
) -> TaskContext:
return TaskContext(
task=task_service
)
resolver.py
コンテキストからサービス情報を取得し、対応する処理を呼び出す実装を記載します。
import strawberry
from web.task.inputs import AddTaskInput, UpdateTaskInput
from web.task.types import TaskType
from strawberry.types import Info
def get_task(id: strawberry.ID, info: Info) -> TaskType:
service = info.context.get_task()
task = service.find(id)
return task
def get_tasks(info: Info) -> list[TaskType]:
service = info.context.get_task()
tasks = service.find_all()
return tasks
def add_task(task_input: AddTaskInput, info: Info) -> TaskType:
service = info.context.get_task()
task = service.create(**task_input.__dict__)
return task
def update_task(task_input: UpdateTaskInput, info: Info) -> TaskType:
service = info.context.get_task()
task = service.update(**task_input.__dict__)
return task
def delete_task(id: strawberry.ID, info: Info) -> TaskType:
service = info.context.get_task()
task = service.delete(id)
return task
router.py
GraphQLのQuery, Mutationの形式を定義し、実行する処理をリゾルバとして渡すように記載します。
import strawberry
from strawberry.fastapi import GraphQLRouter
from resolvers import add_task, delete_task, get_task, get_tasks, update_task
from web.task.types import TaskType
from contexts import get_context
@strawberry.type
class Query:
task: TaskType = strawberry.field(resolver=get_task)
tasks: list[TaskType] = strawberry.field(resolver=get_tasks)
@strawberry.type
class Mutation:
task_add: TaskType = strawberry.field(resolver=add_task)
task_update: TaskType = strawberry.field(resolver=update_task)
task_delete: TaskType = strawberry.field(resolver=delete_task)
schema = strawberry.Schema(query=Query, mutation=Mutation)
task_app = GraphQLRouter(schema, context_getter=get_context)
app.py
from fastapi import FastAPI
from router import task_app
import uvicorn
from database import database_context
api = FastAPI()
def register_controller():
api.include_router(task_app, prefix='/task')
if __name__ == '__main__':
database_context.initialize()
register_controller()
uvicorn.run(app=api, host='0.0.0.0', port=8000)
実装したrouter.pyの内容を登録し、FastAPIを起動する処理を記載します。
起動、API呼び出し
起動コマンドは以下のようになっています。
cd ./src
python -m app
起動に成功するとlocalhost:8000にAPIが立ち上がります。
/taskにGraphQLルータを追加したので、
localhost:8000/taskにブラウザでアクセスするとStrawberryのGraphQL UIページにアクセスできます。
中央のパネルにGraphQLクエリを入力して実行することができます。
左側のメニューからドキュメントを確認したり、クエリの探索的作成もできる便利なUIになっています。
データの投入クエリは以下のようになっています。
mutation addDataSample {
taskAdd(taskInput:{ title:"test", description: "testTask"}){
title
}
}
クエリの実行結果は右側のパネルに表示されます。
投入したデータを一覧取得するクエリを実行してみましょう。
query listtasks {
tasks {
id
title
status
description
updatedAt
}
}
無事、投入したデータを確認することができました。
さいごに
FastAPIとGraphQLライブラリStrawberryを用いて簡単にGraphQLAPIを実装する方法を紹介しました。
ライブラリに則った定義を記載するだけで簡単にGraphQLのAPIを実現できます。
手軽に実現できるため、今後APIを構築する際はGraphQLも選択肢の一つに入るのではないかと思います。
REST API、GraphQL双方の利点、欠点を踏まえながら最適な形式を選択していきたいものですね。
それでは！
Acroquest Technologyでは、キャリア採用を行っています。
- ディープラーニング等を使った自然言語／画像／音声／動画解析の研究開発
- Elasticsearch等を使ったデータ収集／分析／可視化
- マイクロサービス、DevOps、最新のOSSを利用する開発プロジェクト
- 書籍・雑誌等の執筆や、社内外での技術の発信・共有によるエンジニアとしての成長
少しでも上記に興味を持たれた方は、是非以下のページをご覧ください。
www.wantedly.com
