🐧

FastAPI＋GraphQLで非同期処理の基礎を理解する

2023/10/21に公開

GraphQL

概要

今回は初心者が陥りがちな非同期処理について解説します。非同期処理に関する記事はインターネット上に無数にありますが実際にどういう使い方をするのか、どういう仕組みでコードが動いているのかを分かりやすく解説した記事はほとんどありません。そこで今回は初心者がこの記事だけで非同期処理を実践で使えるようにするために世界一わかりやすく書いてみました。対象は初心者の方のため、環境構築から丁寧に解説していきますがもし必要なのなら実際にコードを書いていく工程まで飛ばしてください。

この記事の目標

非同期処理を実践で使えるレベルで理解できる
GraphQLをなんとなく理解できる
データベースの仕組みを少し理解できる

今回使用する技術

今回使用する技術は以下の3つです

FastAPI
MongoDB
GraphQL

今回の非同期処理の実装について

今回やってみる非同期処理は作ったデータが30秒で自動で消えるようにする非同期処理です。非同期処理について簡単に説明するとある処理が実行されている間にもう一つの処理の準備を進めておくことです。通常プログラムは上から下に順番で実行されますが非同期処理を使ったコードは一度実行が分離されて並行して処理してくれるようになります(※少し不適切な表現もありますが今回はわかりやすさのためにかみ砕いた表現をしています)。

データベースの設定

今回使うデータベースはMongoAtlasというクラウドプラットフォームです。NoSQLであるMongoDBを使ってデータの通信を行います.
MongoAtlasのリンクはこちら

サインアップ

MongoAtlasを開いたらまずSign Upからユーザ登録をしてください。Google, GitHub, その他のどのアドレスからでもアクセスできます。

この欄を埋めて登録してください。

デプロイメント

MongoAtlasにログイン出来たら+Createのボタンを押してプロジェクトを開始します。今回はFREEプランでProviderはGoogle Cloudにしておきます(ここでは特に関係ありません)。RegionはTokyoでタグの設定はしなくていいです。NameをCluster0で作成します。

Security Quickstartの設定です。
UsernameとPasswordは自由に設定してください。ただ後でこのPasswordとusernameは使用するので必ず覚えておいてください。確定したらCreate Userをクリックしてください。環境はMy Local EnvironmentでIPアドレスは0.0.0.0/0を使用してください。このアドレスはワイルドカードといってすべてのIPアドレスに対応しています。実務の開発ではセキュリティが脆弱になるので本番での使用は極力避けましょう。そして設定を完了してください。

データベースを作成

このページがホーム画面です。DeploymentのDatabaseを選択しBrowse Collectionにいきます。そしてOwn Dataで作成を選択して下さい。今回はDatabase_nameをFastAPI、Collection_nameをUserとします(他の設定は必要ないです)。
これで一度MongoAtlasの設定は完了です。

FastAPIのプロジェクト作成

次にFastAPIでプロジェクトを作成します。FastAPIについて簡単に説明しますと、Pythonのwebバックエンドフレームワークの一つで、軽量でNode.jsやGo言語に匹敵する超高速な通信速度が特徴です。今回はVScodeを使って書いていきます。(※Windowsを使用しているのでMacユーザの場合は少し書き方が変わる可能性があります)

プロジェクトを作成

mkdir async_py_pj

ターミナルでフォルダを作成します。その後VScodeでフォルダを開いてください。
次に仮想環境のセッティングをします。ターミナルで以下のコマンドを入力します

python -m venv venv

venv\Scripts\activate

これで仮想環境の構築ができたのでここにライブラリなどをインストールする作業を行います

pip install fastapi uvicorn[standard] sqlalchemy strawberry-graphql
pip install pymongo
pip install asyncio
pip install python-dotenv

を実行して必要なパッケージをインストールしてください(もし容量が少ないのであれば仮想環境での実行をお勧めします。今回は直書きしていきます）。
インストールしたパッケージについて少し解説すると

fastapi:FastAPIの実行環境
uvicorn:アプリケーションサーバー
sqlalchemy:データベースと通信する用のライブラリ
strawberry-graphql:GraphQL APIを使用するためのライブラリ
pymongo:MongoDBと接続するためのライブラリ
asyncio:非同期タスクを設定するためのライブラリ
になります。これらのフレームワークの内部の仕組みを知る必要はありませんが適応範囲や実行の水準やレベル感などは理解しておきましょう。

main.py

#必要なライブラリをインポート
import os
import strawberry
from fastapi import FastAPI
from strawberry.asgi import GraphQL
from pymongo import MongoClient
from fastapi import FastAPI, BackgroundTasks, Depends
import asyncio
from dotenv import load_dotenv
from typing import List

以上のパッケージをmain.pyファイルを作成しimportしてください。

MongoDBとFastAPIを連携させる

次にMongoDBと連携する操作を実行します。MongoAtlasにもどってOverViewからConnectを選択します。Driversを開いてDriverをPython、versionは3.12 or laterを選択します。pymongoは先ほどインストールしたので必要ありません。

mongodb+srv://your_username:<password>@cluster0.vnobe3a.mongodb.net/?retryWrites=true&w=majority

をコピーして.envファイルを作成し書き込みます。ここでyour_usernameとpasswordがあなたがMongoAtlasで設定したものであることに注意してください。

.env

MONGO_URL = 'mongodb+srv://ritsumeitaro:<password>@cluster0.vnobe3a.mongodb.net/?retryWrites=true&w=majority&appName=Myapp'

main.pyにMONGO_URLを定義しない理由はGitHubなどで本番環境にデプロイするときにセキュリティの対策を行うためです。今回はGitHubに上げるところは言及しないので知らない方はほかの記事を利用してください。

最後にmain.pyに戻って以下のコードを先ほどインストールしたコードの下に続けて書きます。

main.py

#.envファイルを読み込む
load_dotenv()

#MONGO_URLにアクセスしてルーム名'FastAPI'を呼び出す
client = MongoClient(os.environ["MONGO_URL"])
db = client["FastAPI"] #設定したDatabase名
collection = db["User"] #設定したCollection名

これでMongoDBとFastAPIをつなげることができました。

GraphQLを使ってデータを出力してみる

main.pyに続けて以下のコードを入力します

main.py

# スキーマの型を定義する
@strawberry.type
class User:
    id: int
    name: str
    age: int

# クエリの定義をする
@strawberry.type
class Query:
    @strawberry.field
    def user(self) -> User:
        return User(id=1, name="Taro", age=19)

# スキーマを作成する
schema = strawberry.Schema(query=Query)

#GraphQLエンドポイントを作成する
graphql_app = GraphQL(schema)

# FastAPIアプリのインスタンスを作る
app = FastAPI()

#/graphqlでGraphQL APIへアクセスできるようにし、適切なレスポンスを出力
app.add_route("/graphql", graphql_app)

ここではUser型で返してほしい情報を定義します。今回はユーザ情報が知りたいのでidとnameとageを指定しています。
ここで@strawberry.はデコレータといいstrawberryのライブラリを呼び出しています。
クエリの定義ではデータを返す処理を書いています。クエリとは日本語で取り決めや命令という意味でデータベースに操作を指定する仕組みのことです。今は決め打ちで入力されたid name ageを返すように指定してあります。

一度これでGraphQLのエンドポイントにアクセスしてみましょう。
ターミナルで以下のコマンドを実行してください。

uvicorn main:app --reload

/graphqlにアクセスしてみます

左のようにクエリを入力して右の結果がでれば成功です。これで自分が定義したデータを出力することができました

GraphQLでデータベースと通信してみる

今度は決め打ちでなく実際に入力した情報を返すように処理を書いてみます。データベースに情報を送る(書き込む、更新する)操作をMutationといい、データベースから情報を引っ張ってくる逆の操作をQueryといいます(ざっくり)。書き込みたい情報を定義するときには@strawberry.inputというデコレータを使用します。

main.py

@strawberry.type
class User:
    id: int
    name: str
    age: int
  
# 入力データの定義をする（追加）  
@strawberry.input
class Register:
    id: int
    name: str
    age: int

@strawberry.type
class Query:
    @strawberry.field
    def user(self) -> User:
        return User(id=1, name="Taro", age=19)

# Mutationを定義する（追加）
@strawberry.type
class Mutation:
    @strawberry.field
    def register(self, regist:Register) -> User:
        collection.insert_one(regist.__dict__)
        return regist

Mutationで書かれている内容を解説します。クラス内のregisterメソッドは上で定義したRegister型のデータをMutationで渡すことでMongoDbがUser型として返してくれます。ここでRegisterとUserは同じ要素で構成されているのでメソッド内での変換や入力を増やす操作は不要です。
nsert_oneメソッドを使ってcollectionにRegisterで入力された情報を辞書型でデータベースに格納します。ここでコレクションは最初に作ったUserテーブルのことです。
これで処理が完成したのでもう一度サーバーを立ち上げてみます。
左のようにMutationを入力すると右ターミナルに実行結果が返ってきます。入力した情報が反映されていれば成功です。

では実際にデータベースが更新されているかMongoAtlasに見に行きましょう。

しっかり新しいユーザ情報が登録されていますね(新しいデータが反映されていないという人は右上のrefreshボタンを押してみてください)。これでサーバとの通信が完了です。

ここまでが環境のセッティングでした。皆さんお疲れさまでした。ここまでだけでいろいろな情報を詰め込んだので頭が痛いと思います。少し休憩してから再開してくださいね。

非同期処理を実装してみる

今回の本題である非同期処理の実装をしてみます。今回実装する処理は時間制限でユーザ情報が自動で消滅するといった処理です。これはかなり実践的な仕組みで例えばその場限りで使うようなアプリの会員登録や一時的にルームを作成して少人数のスペースをつくったりするなどのときに時間制限をつけておくことでデータベースに情報がたまらなくなりサーバーへの負担が軽くなります。
会社レベルで使えるテクニックでこれをそのまま職場で使うことも可能です。

main.py

#ユーザを削除するための非同期関数を定義
async def schedule_user_deletion(user_id):
    #30秒間スリープする処理
    await asyncio.sleep(30)
    #ユーザを削除
    collection.delete_one({"user_id": user_id})
    
#非同期関数として定義し直す
strawberry.type
class Mutation:
    @strawberry.field
    async def register(self, regist:Register) -> User:
        collection.insert_one(regist.__dict__)
	#非同期タスクとして以下の処理を予約
        asyncio.create_task(schedule_user_deletion(regist.user_id))
        return regist

ここで非同期処理の仕組みについて解説します。以下は非同期処理のフローチャートです。nコの処理に関して実行順序がk番目である処理kについて考えてみます。非同期タスクを受け取るまでは実行順序を遵守してプログラムを読んでいきます。処理kに到達した後まず処理kは非同期タスクを読み込みます。そして非同期処理の実行順序から外れた位置で処理を待機させます。その間に処理k+1に進みます。非同期処理は処理k+1に並行して処理を実行します(並行処理は一つずつ処理を切り替えるので別物です)。非同期処理が実行し終わるとメインの処理に遷移して実行順序は一位にもどります。難しい内容なので読み飛ばしても理解できるように書きました。

今回のコードではcollection.insert_oneの実行が終わった後にasyncioで非同期処理する予定のタスクを読み込んでメインの処理で実行せずに次の処理であるreturn registに遷移します。ここでメインの処理は終了しますが非同期処理は実行されていてここでは上で定義したschedule_user_deletion関数が実行されている状態です。

まだ非同期処理をさっぱりわからんという人も大丈夫です。今から実際の処理を解説します。

これが非同期処理を使わないコードフローです。処理kが起きた後順番通りに非同期タスクに処理が移り非同期タスクが終わるまで処理k+1にはいけません。ここで問題が起きます。今回の非同期タスクは30まってからユーザのデータを消すという処理です。つまり前の処理でRegisterで登録したデータを30秒待ったあと消してそれを返すという処理になります。つまり何も返ってきません(データを消してるので当たり前ですね)。
これで非同期処理とそうでない処理について仕組みを理解できたと思います。ではこのコードを実際に実行して確かめてみましょう。

まずGraphQLからデータを更新します。データの保存は成功したようです。次にMongoAtlasへデータを確認しに行きます。

しっかり新しいデータが保存されていますね。これで30秒待ちます。

しっかり二つ目のデータが消えています。データの自動削除はこれで完成です。

まとめ

今回は初心者が詰むプログラミング処理ランキング一位の非同期処理についてハンズオン形式で解説しました。自分の性格上しっかり０から解説してくれるものが好きなので今回は環境構築から解説してみました。何にでもいえることなのですがこれをただ読むだけでは正直時間の無駄です。自分でいじって頭をつかってアウトプットして初めて使える技術になります。オススメは自分でデータ型や非同期タスクを変えてみることです。この記事を読んでプログラミング力が向上してくれたらうれしいです。
ほなまた

概要

この記事の目標

今回使用する技術

今回の非同期処理の実装について

データベースの設定

サインアップ

デプロイメント

データベースを作成

FastAPIのプロジェクト作成

プロジェクトを作成

MongoDBとFastAPIを連携させる

GraphQLを使ってデータを出力してみる

GraphQLでデータベースと通信してみる

非同期処理を実装してみる

まとめ

Discussion