FastAPI 入門

この記事では、FastAPI を実装した話について書いています。この記事では、FastAPI をDocker を利用した環境構築から、アプリを構築する方法を書いています。

FastAPI とは

FastAPI は、Django やFlask といったPython のWebフレームワークの一つです。

このフレームワークには、次のような特徴があります。

OpenAPI に基づいて、自動的にJSON Schema モデルを生成してくれます
Python のASGI フレームワークであるUvicornにより、 Node.jsやGo言語並のパフォーマンスが利用できます
Pydantic を利用して、モデルの型やバリデーションを定義することができます
API を定義すると、Swagger UI,Redoc によるドキュメントが自動生成されます
GraphQLやWebSocketも対応しています

環境構築

このFastAPI は、Python のバージョンが3.6 以上であるという条件があります。今回、Docker を利用しての環境構築を行っていきたいと思います。

まず、requirement.txt を書いていきます。

requirement.txt

fastapi
sqlalchemy 
uvicorn 
email-validator
graphene 
jinja2
aiofiles
PyMySQL

続いて、Dockerfile を次のように書いていきます。

FROM python:3.8.0-alpine
WORKDIR /api
# gcc ,openssl をインストールしなければ、python のセットアップ時にエラーを起こしてしまう。
RUN apk add build-base libffi-dev
COPY ./requirement.txt .
RUN pip install --upgrade setuptools && pip install -r requirement.txt
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

この時、--host 0:0:0:0 を設定しなければ、コンテナが起動できても、ローカルホストへアクセスできません。 Flask や Rails などのフレームワークでDocker で起動したにもかかわらずlocalhost へアクセスできない場合は、 起動ホストを'0.0.0.0' と設定して起動してみてください。

次に、docker-compose.yml を利用して構築していきます。

docker-compose.yaml

version: '3.0'
services:
  api:
    restart: always
    build: ./api
    container_name: 'api'
    ports:
      - 8000:8000
    volumes: 
      - ./api:/api
    depends_on:
      - db
  db:
    restart: always
    image: mariadb:latest
    ports: 
      - 3306:3306
    container_name: 'api_db'
    environment: 
      MYSQL_DATABASE: example
      MYSQL_ROOT_PASSWORD: example
      TZ: 'Asia/Tokyo'
    volumes: 
      - ./docker/db/data:/var/lib/mysql
      - ./docker/db/my.cnf:/etc/mysql/conf.d/my.cnf
      - ./docker/db/sql:/docker-entrypoint-initdb.d

Hello World

まず、このフレームワークを実装して "Hello World" と返すよう実装します。

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

この時、async def (関数)として、非同期関数で実装するほうががいいとされます。このことについては、公式ドキュメントを参照してください。

実行するには、次のようにします。

$ docker-compose build
$ docker-compose up

起動したら、cURL を利用してhttp://localhost:8000 へアクセスすると,

$curl localhost:8000
{"message":"Hello World"}

とJSON 形式のデータが返ってきます.

このとき、生成されたドキュメントを確認するにはこちらです。https://localhost:8000/docs

パス、クエリのパラメーターの取得

次に, Path のパラメーターやクエリの値を取得するには、次のように取得します.

from fastapi import FastAPI,Query


app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.get("/{id}")
async def getId(id: int):
    return {'id':id}

@app.get("/items/")
async def read_items(q: List[str] = Query(None)):
    query_items = {"q": q}
    return query_items

curl を利用して、返ってくるJSON データを確認すると、次の通りになります。

$curl localhost:8000/2
{"id":2}

$curl localhost:8000/items/?q=20
{"q":20}

# Validation Error　は次のように返します
$curl localhost:8000/hello
{
  "detail":
   [
    {
      "loc":
      [
        "path",
        "id"
      ],
      "msg":"value is not a valid integer",
      "type":"type_error.integer"
    }
  ]
}

"""
- loc
  どこでエラーが発生しているかをリスト型で伝えてくれます。
  先頭の要素はエラー箇所を、その次の要素は、ネスト状になったエラー箇所の場所を示してくれます。
- type
  エラーの種類を示します
- msg
  エラー理由を説明してくれます。

"""

Request, Response の取得

Request のBody について、pydantic を利用して、型安全に取り出すことができます。

from fastapi import FastAPI
from pydantic import BaseModel
from pydantic.types import EmailStr

app = FastAPI()


class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: str = None


class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: str = None


@app.post("/user/", response_model=UserOut) # Response の型を定義
async def create_user(*, user: UserIn):     # Request の型を定義
    return user

この時、Resonseの型を定義する際は、@app.post('/',resonse_model=(モデルの型)) というようにデコレーターの引数であるresponse_modelに型を定義します。

HTML テンプレート (Jinja2) を利用する

Flask で標準で用いられていたJinja　を利用して、HTML ファイルを返すこともできます。しかし、それらを利用するためには、jinja2を事前にインストールする必要があります。

また、CSS やJS といった静的ファイルを利用する際には、aiofiles をインストールします。

from fastapi import FastAPI
from starlette.requests import Request
from starlette.staticfiles import StaticFiles
from starlette.templating import Jinja2Templates

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")


templates = Jinja2Templates(directory="templates")


@app.get("/items/{id}")
async def read_item(request: Request, id: str):
    return templates.TemplateResponse("item.html", {"request": request, "id": id})

item.html

<html>
<head>
    <title>Item Details</title>
    <link href="{{ url_for('static', path='/styles.css') }}" rel="stylesheet">
</head>
<body>
    <h1>Item ID: {{ id }}</h1>
</body>
</html>

styles.css

h1 {
    color: green;
}

SQL を扱う

Web アプリ制作でSQLを扱う際、データベースに直接コマンドを書いて実行するといったことは非常に大変です。そのため、ORM Wrapper を通してDB の設計を行います。Python では、SQLALchemy を通して設計します。今回は、SQLAlchemy とpydantic を使い、型安全なデータベースを作っていきます。

詳しい使い方については、SQLAlchemy の公式ドキュメントを参照してください。

from fastapi imoprt FastAPI
from pydantic import BaseModel

# DB へのURL 
SQLALCHEMY_DATABASE_URL = "mariadb://db:3306/example

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

def get_db():
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()


class Account(BaseModel):
    id: Int
    name: str
    password: str
    email: str 

app=FastAPI()

@app.get('/')
async def findAllAccount():
  return 
   
@app.post('/')
async def saveAccount():
    pass

GraphQL、WebSocket の実装

GraphQL, Websocket を利用する際には、次のようにします。

GraphQL

import graphene
from fastapi import FastAPI
from starlette.graphql import GraphQLApp


class Query(graphene.ObjectType):
    hello = graphene.String(name=graphene.String(default_value="stranger"))

    def resolve_hello(self, info, name):
        return "Hello " + name


app = FastAPI()
app.add_route("/", GraphQLApp(schema=graphene.Schema(query=Query)))

WebSocket(Jinja2 を組み合わせて構築してみる)

from fastapi import Cookie, Depends, FastAPI, Header
from starlette.responses import HTMLResponse
from starlette.status import WS_1008_POLICY_VIOLATION
from starlette.websockets import WebSocket

app = FastAPI()

html = """
<!DOCTYPE html>
<html>
   <head>
       <title>Chat</title>
   </head>
   <body>
       <h1>WebSocket Chat</h1>
       <form action="" onsubmit="sendMessage(event)">
           <label>Item ID: <input type="text" id="itemId" autocomplete="off" value="foo"/></label>
           <button onclick="connect(event)">Connect</button>
           <br>
           <label>Message: <input type="text" id="messageText" autocomplete="off"/></label>
           <button>Send</button>
       </form>
       <ul id='messages'>
       </ul>
       <script>
       var ws = null;
           function connect(event) {
               var input = document.getElementById("itemId")
               ws = new WebSocket("ws://localhost:8000/items/" + input.value + "/ws");
               ws.onmessage = function(event) {
                   var messages = document.getElementById('messages')
                   var message = document.createElement('li')
                   var content = document.createTextNode(event.data)
                   message.appendChild(content)
                   messages.appendChild(message)
               };
           }
           function sendMessage(event) {
               var input = document.getElementById("messageText")
               ws.send(input.value)
               input.value = ''
               event.preventDefault()
           }
       </script>
   </body>
</html>
"""

@app.get("/")
async def get():
   return HTMLResponse(html)


async def get_cookie_or_client(
   websocket: WebSocket, session: str = Cookie(None), x_client: str = Header(None)
):
   if session is None and x_client is None:
       await websocket.close(code=WS_1008_POLICY_VIOLATION)
   return session or x_client


@app.websocket("/items/{item_id}/ws")
async def websocket_endpoint(
   websocket: WebSocket,
   item_id: int,
   q: str = None,
   cookie_or_client: str = Depends(get_cookie_or_client),
):
   await websocket.accept()
   while True:
       data = await websocket.receive_text()
       await websocket.send_text(
           f"Session Cookie or X-Client Header value is: {cookie_or_client}"
       )
       if q is not None:
           await websocket.send_text(f"Query parameter q is: {q}")
       await websocket.send_text(f"Message text was: {data}, for item ID: {item_id}")

最後に

これまで、PythonでWebアプリを作成するはあまり好まれるものではありませんでした。これまでのFlaskやDjango といったものは、Node.js やGo 言語に比べ、速度が遅いという問題がありした。マイクロサービスアーキテクチャが普及していく中で、速度かつ軽量なフレームワークが好まれるために、処理速度の遅さは致命的な問題でした。

しかしながら、これらの条件を満たしたこのFastAPIを利用すれば、より高速なレスポンスによるパフォーマンスの向上が期待できます。

さらに、Pydantic による型安全が行えるため、開発する際の手助けにもなるので、今後Flaskの代替フレームワークとして、ますます普及が進むと思われます。

FastAPI 入門

FastAPI とは

環境構築

Hello World

パス、クエリのパラメーターの取得

Request, Response の取得

HTML テンプレート (Jinja2) を利用する

SQL を扱う

GraphQL、WebSocket の実装

GraphQL

WebSocket(Jinja2 を組み合わせて構築してみる)

最後に

参考文献

タグ一覧

最新記事

Web アプリ開発を効率よく学ぶには

Dockerfile にBcrypt を導入する方法

Gridsome でブログを制作した話

FastAPI 入門

エンジニア職インターンシップ選考に対してのアドバイス

Home

Blog

FastAPI 入門

FastAPI とは

環境構築

Hello World

パス、クエリのパラメーターの取得

Request, Response の取得

HTML テンプレート (Jinja2) を利用する

SQL を扱う

GraphQL、WebSocket の実装

GraphQL

WebSocket(Jinja2 を組み合わせて構築してみる)

最後に

参考文献

タグ一覧

最新記事

Web アプリ開発を効率よく学ぶには

Dockerfile にBcrypt を 導入する方法

Gridsome で ブログを制作した話

FastAPI 入門

エンジニア職インターンシップ選考に対してのアドバイス

Dockerfile にBcrypt を導入する方法

Gridsome でブログを制作した話