はじめに
最近アクアリウムを始めました、菅野です。
プログラムと異なり、生体を扱う都合上の想定外を楽しみながら試行錯誤しております。
さて、皆さんは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テーブルの定義 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 = list(cls._store.values()) 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()
domainディレクトリ
内部で行うビジネスロジック(今回では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)
web/taskディレクトリ
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
srcディレクトリ直下
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では、キャリア採用を行っています。少しでも上記に興味を持たれた方は、是非以下のページをご覧ください。 www.wantedly.com
- ディープラーニング等を使った自然言語/画像/音声/動画解析の研究開発
- Elasticsearch等を使ったデータ収集/分析/可視化
- マイクロサービス、DevOps、最新のOSSを利用する開発プロジェクト
- 書籍・雑誌等の執筆や、社内外での技術の発信・共有によるエンジニアとしての成長