Taste of Tech Topics

Taste of Tech Topics

Acroquest Technology株式会社のエンジニアが書く技術ブログ

FastAPI+StrawberryでGraphQLのAPIを実現する

はじめに

最近アクアリウムを始めました、菅野です。
プログラムと異なり、生体を扱う都合上の想定外を楽しみながら試行錯誤しております。

さて、皆さんはAPIサーバを構築する際に、どのAPI形式を用いていますか?
まだまだREST形式で実装することが多いかとは思いますが、
GraphQLを用いることも増えてきているのではないでしょうか?

今回は、そんなGraphQLをFastAPIと各種ライブラリを用いて簡単に実装する方法を紹介していこうと思います。

GraphQLとは

GraphQLは、Meta社(旧Facebook社)によって開発・公開されたAPI仕様です。クエリ形式で、処理やパラメータの内容を指定します。
RESTとの比較としては、

  • クライアント側で取得したい情報をクエリとして渡すことができるため、 利用しないデータを無駄に受け取らなくて済む。
  • 一つのエンドポイントに対し複数リソースのクエリを一度にリクエストできるので、APIコール数を削減できる

等の利点があります。

GraphQL | A query language for your API

構成

PythonAPI作成フレームワーク、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では、キャリア採用を行っています。
  • ディープラーニング等を使った自然言語/画像/音声/動画解析の研究開発
  • Elasticsearch等を使ったデータ収集/分析/可視化
  • マイクロサービス、DevOps、最新のOSSを利用する開発プロジェクト
  • 書籍・雑誌等の執筆や、社内外での技術の発信・共有によるエンジニアとしての成長
  少しでも上記に興味を持たれた方は、是非以下のページをご覧ください。 www.wantedly.com