💨

FastAPI と Mangum で作るサーバーレスAPI

2023/02/04に公開

AWS

はじめに

こんにちは。hayata-yamamoto です。

唐突ですが、皆さんは「サーバーレス」お好きでしょうか？AWS であれば、Lambda, Fargate, Aurora Serverless あたりを思い浮かべるかもしれません。GCP にも似たようなサーバーレスサービスがありますよね。サーバーを自分たちで用意せず、クラウドベンダーが”いい感じに”設定したサーバーを拝借できることで、エンジニアはアプリケーションの開発に集中できる、そんなサービスです。オートスケールの設定が容易にできたり、リソース使用にかかる費用が従量課金になっており、ワークロードに合わせてコスト最適化を行いやすかったりと、さまざまなメリットがあります。

今回は、弊社でも採用しているサーバーレス構成（ServerlessFramework, FastAPI, Mangum)を紹介しながら、以下の点についてお伝えします。

どうしてこれらの技術を使用しているのか？
どうやって使うのか？
実際に使用してみてどうか？
どういう人にお勧めできるのか？
どう導入していったらいいのか？

どうしてこれらの技術を使用しているのか？

端的に言うと以下の二点を実現するためです。

ローカルでの検証を容易にするため
型ヒントによる恩恵を享受し、余計な実装コストを削減するため

以下では詳しく、技術選定や検討したことの経緯をご紹介します。

背景

弊社では、サーバーレスサービスを中心にしたバックエンドの提供を行っています。AppSync で提供している GraphQL と、APIGateway で提供している REST API が共存する状況となっており、Serverless Framework では主に以下の２種類の Lambda 関数を提供しています。

AppSync のデータソースとして実行される Lambda 関数
APIGateway で利用される Lambda 関数

当初、AppSync を中心にしてバックエンドの API を提供してきた弊社では、Lambda 関数は主に、 AppSync のデータソースとして利用されてきました。しかし、AppSync を中心としたバックエンドの提供にはさまざまな辛さがあり、私が参画して以降全面的に GraphQL から REST への移行を進めています。現在では APIGateway を中心に REST で API を提供しており、FastAPI の生成する OpenAPI の UI を用いつつ、ローカルで検証しながら開発するスタイルに移行しました。

以前は、ローカルでの検証が難しかった

「ん？それはサーバーサイドの開発であれば、当然なのでは？」

と思った人もいるかもしれません。まさにその通りです。恥ずかしい話ですが、サーバーレスの技術を採用していることと、AppSync と Lambda の依存関係があることが影響し、ローカルでのテストを行うコストが大変高い状況に以前はありました。AppSync から投げ込まれるリクエストが何かを知った上で、Lambda を書くという相互依存のあるシステムだったのです。

一方、APIGateway の方もそれなりに辛さがありました。Serverless Framework にも serverless-offlineというプラグインがあり、ローカルでの起動はできるのですが、Lambda で処理を実装する際には APIGateway から投げ込まれるイベントやコンテキストの形式を知った上で書かなくてはならず、さらに単体テストを書く際にはそのデータ形式を再現する必要がありました。AWS Powertools が Event Source Data Classes という AWS のイベントをまとめたデータクラスを提供してくれてはいるものの、気休め程度にしかなりませんでした。

それらが積み重なり、開発チームでは「FastAPI であれば、リクエストのオブジェクトを明示的に記載した上で OpenAPI から JSON を送りつけるだけでテストできるのに...」と考えていました。

型ヒントの恩恵を十分に受けるため

弊社の Python プロジェクトでは、最近のモダンな Python の流れに合わせて型の記述を行いながら開発を進めています。関数の入出力に対して型を書くことはもちろん、Pydantic を用いたデータバリデーションと設定管理を行っています。

型ヒントと pydantic の恩恵を十分に受けるため、以前は APIGateway でイベントを受け取る際には、関数の一番最初で Pydantic を用いたイベントのバリデーションを実行し、問題なければ後続のロジックを実行するという実装をしていました。具体的には、以下のようなイメージです。

from pydantic import BaseModel, ValidationError
from logging import getLogger

logger = getLogger(__name__)


class TestRequestPathParameters(BaseModel):
    user_id: int


class TestRequestQueryStringParameters(BaseModel):
    text: str


class TestRequest(BaseModel):
    pathParameters: TestRequestPathParameters
    queryStringParameters: Optional[TestRequestQueryStringParameters] = None


# GET /something/:user_id?text=string の形式を想定
def handler(event: Dict[str, Any], _: Any) -> None:
    logger.info(event)
    try:
        event = TestRequest.parse_obj(event)
    except ValidationError:
        logger.exception("Failed to validate event", extra={"event": event}) // Sentry に通知が飛ぶ
        return {"statusCode": 422, "message": "validation error"}

    # ハンドラの処理が続く
    ...

しかしよく考えると、Pydantic のモデルを用いてリクエストオブジェクトを検証する部分は、FastAPI であればフレームワークで勝手にやってくれる部分です。実装を加える度、「なんでこれ書かなきゃいけないんだろう...」と感じていました。本質的には以下で十分なものに、込み入った実装をしているなと感じていたのです。

from pydantic import BaseModel, ValidationError
from fastapi import FastAPI
from logging import getLogger

logger = getLogger(__name__)
app = FastAPI()


@app.get("/something/{user_id}")
def handler(user_id: int, text: Optional[str] = None) -> None:
    logger.info(event)

    # ハンドラの処理が続く
    ...

上記 2 つの理由を考慮し、 FastAPI + Mangum を用いて API を提供していくことになりました。

どうやって使うのか？

最初に、必要なライブラリを pip や poetry などでインストールしておきましょう。ServerlessFramework の方は、Serverless Python Requirements を入れておくと良いです。使用するパッケージ管理ツールによって設定が異なりますので、ドキュメントを確認しながら serverless.yml に設定追加を行ってください。

pip3 install fastapi uvicorn mangum
poetry add fastapi uvicorn mangum

ServerlessFramework で FastAPI + Mangum を使用するためには、最低以下の二つのファイルに手を加える必要があります。以下に、1 例を示しておきますが、実装例は、参考にある記事のほうが詳しく記載されています。さまざまな実装方法があるようですので、どの取り入れ方が自社にマッチするかはいくつか記事を読みながら判断されると良いです。

serverless.yml
app.py

serverless.yml

functions:
  index:
    handler: xxxxxxx # FastAPI のアプリケーションが宣言されているファイルまでのパスを書く
    events:
      - http:
          path: /{proxy+}
          method: ANY

app.py

from fastapi import FastAPI
from mangum import Mangum

app = FastAPI()


@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

handler = Mangum(app)

FastAPI の実装ができたらローカルで起動し、http://localhost:8000/docs を確認しましょう。問題なければ、sls deploy して APIGateway 経由でもレスポンスが帰ってくることを確認してください。

uvicorn <app.pyのパス>:app

実際に使用してみてどうか？

今のところ、不都合を感じることはありません。メリット・デメリットを簡単に整理すると以下のようになります。

FastAPI + Mangum	メリット	デメリット
あり	・ローカルでテスト用サーバーが起動できる。・単一のエンドポイントを使えるため、コールドスタートを感じにくい・コンテナ化して他のサービスに移管しやすい	・CloudWatch のロググループが一つにまとまってしまう。・FastAPI の知識が必要
なし	・手軽に始められる・処理内容に応じて、実装とログをが作成されるため、単一の機能のログを追いやすい	・ローカル環境でのテストがめんどくさい。・各関数の初期起動時はコールドスタートが発生する大量の Lambda 関数ができてしまい、関数を横断するログを追うのが大変

どういう人にお勧めできるのか？

Python + サーバーレスで API を提供している人（コンテナでサービス提供はしていない）
ある程度、API の規模が大きくなってきており、チームも複数人が同時に API を開発するようになっている人
コンテナにしたいけど、そこまでするにはコストが高い。しかし、もっと手軽に API の開発ができるようになって欲しいし、開発効率をあげたいとおもっている人

どう導入していったらいいのか？

新機能から順に FastAPI + Mangum に移行していくことをおすすめします。順番を整理すると、

対象とする新機能を選定する
→ なるべくシンプルな実装で、影響範囲の小さいものが望ましい
新機能に FastAPI + Mangum を導入し、パフォーマンスやログの吐かれ方などを確認する
→ ログ収集基盤などを有しており、ログの吐き出し形式が決まっている場合は調整したほうがよい
プロダクション利用できるかどうか、自社のシステム基準に合うかどうかを判断する
→ システムに求められる可用性や、想定リクエスト数などを鑑みて判断する
新機能にいくつかエンドポイントを作成してみて、Lambda リソースの使用状況をモニタリングする
→ 複数の機能のログが一箇所に出力されるため、ログフィルタなどが必要なら追加する
問題ないと判断できれば、既存のエンドポイントも FastAPI で置き換えていく
→ 新しいバージョンのエンドポイントを作成する
全面的にロールアウトする
→ クライアント側の API コール先を新バージョンに向けていく

図にまとめてみると、

終わりに

トドケールでは、一緒に働いてくれるエンジニアを募集しています!