🏹

FastAPIで学ぶPythonによるREST API開発の基本

2023/05/02に公開1件

Python

API

FastAPI

tech

はじめに

今回の記事では、FastAPIでREST APIを開発する手順を簡潔に解説する。

本記事の対象読者

Pythonの基本文法(データ型、条件分岐、繰り返し)を理解している人
RailsやLaravel等のWebフレームワークで簡単なWebアプリケーションを開発できる人
FastAPIで簡潔にREST APIを開発したい人

用語解説

FastAPI

FastAPIの公式ドキュメントによると、以下のように説明されている。

FastAPI is a modern, fast (high-performance), web framework for building APIs with Python 3.7+ based on standard Python type hints.

簡潔に言えば、FastAPIはPythonでAPIを開発するために開発されたWebフレームワークである。PythonでAPIを開発できるフレームワークにはFlaskやDjangoがあるが、型定義に基づいたAPIを開発できるのが最大の違いとして考慮される。

uvicorn

uvicorn（アッバーコーン）は、Python製のASGI（Asynchronous Server Gateway Interface）サーバーだ。ASGIは、非同期処理をサポートするWebアプリケーションフレームワークに対する標準的なインターフェースだ。

FastAPIのようなASGIフレームワークで開発されたWebアプリケーションを実行するために、uvicornを使用することが一般的だ。uvicornは、非同期処理に特化しており、高速でスケーラブルなWebアプリケーションを実現することができる。

具体的には、uvicornはASGIフレームワークから受け取ったHTTPリクエストを非同期処理で処理し、HTTPレスポンスを返す。また、uvicornは、HTTP/1.1およびHTTP/2プロトコルをサポートしており、TLS/SSLの暗号化通信もサポートしている。

FastAPIの場合、以下のようにコマンドラインからuvicornを起動することで、Webアプリケーションを実行できる。

uvicorn main:app --reload

ここで、mainはFastAPIアプリケーションが定義されたPythonファイルの名前であり、appはFastAPIアプリケーションのインスタンスだ。--reloadは、コード変更を検出した際にサーバーを再起動するオプションだ。

簡単に言えば、uvicornは、FastAPIなどのASGIフレームワークで開発されたWebアプリケーションを高速で安定して実行するためのサーバーだ。

SQLite

SQLiteは、軽量で埋め込み型のリレーショナルデータベース管理システムである。標準のSQL言語をサポートし、トランザクション処理やACIDプロパティの保証、多数のテーブルやインデックス、トリガー、ビュー、外部キー制約などの機能を持っている。

また、SQLiteは多くのプログラムに簡単に組み込むことができる。SQLiteは、Webブラウザやスマートフォンなど、多くのシステムで広く利用されており、簡単に導入できるのが最大の強みである。

pydantic

pydantic(読み：パイダンティック)とは、Pythonのデータ検証ライブラリで、主にデータモデリングやAPI開発に利用される。pydanticは、データの検証、変換、シリアル化をサポートし、Pythonの型ヒントを活用して型安全なコードを書くことができる。

FastAPIでは、HTTPリクエストのボディやクエリパラメータ等のデータを、pydanticで検証できる。また、データベースから取得したデータをpydanticにマッピングできる。

簡潔に述べると、pydanticライブラリを活用すれば、動的型付け言語PythonでもTypeScriptのような開発ができる。

実際の手順

(1) FastAPIのインストール

pip install uvicorn[standard] fastapi[all]

(2) APIのエンドポイントを定義する

main.pyを作成して、以下のようにプログラムを書く。

main.py

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

以上のようにFastAPIオブジェクトを作成し、@app.get()デコレータを使って記述する。例えば、以下のようなコードで/items/{item_id}というエンドポイントを作成する。

(3) APIのエンドポイントにアクセスする

APIのエンドポイントにアクセスするためには、HTTPリクエストを送信する。上述の例では、GETメソッドでitems/10にアクセスすると以下のようなレスポンスが返される。

curl "http://localhost:8000/items/10/"

以上のコマンドを入力して実行すると、以下のようなレスポンスが出力される。

{"item_id": 42}

(4) `main.py`でFastAPIのアプリサーバを構築する

main.pyを以下のように編集する。

main.py

from fastapi import FastAPI

app = FastAPI()

# 以下のコードを追加する
@app.get("/")
async def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None): # 第二引数にqを設定する
    return {"item_id": item_id, "q": q}

APIにアクセスするには、Webブラウザのアドレスバーにhttp://localhost:8000/items/42?q=testと入力するか、curl等のコマンドラインツールを活用する。

以下のようなJSONレスポンスが出力される。

{"item_id": 42, "q": "test"}

以上が、FastAPIでREST APIを開発するための一般的な手順となる。

タスクアプリのREST APIを開発する

本章では、FastAPIを使った簡単なタスクアプリを開発する手順を簡潔に解説する。そこで、Pythonで型定義を使った開発ができるライブラリpydanticを使う。

手順

FastAPIやuvicronは予めインストールされていることを前提に手順を説明する。

(1) `model.py`の新規作成

model.py

from pydantic import BaseModel

class Task(BaseModel):
    title: str
    description: str = None
    completed: bool = False

pydanticライブラリを活用するため、pydantic.BaseModelを継承したTaskモデルを定義する。

Pythonでpydanticを活用して変数に型定義を書く際には、以下のようにコードを書く。

class Task(BaseModel):
    title: str
    description: str = None
    completed: bool = False

ここで余談になるが、型定義と聞いて、TypeScriptの型定義を連想する人も少なくないだろう。たとえば、TypeScriptの場合は以下のように、関数であれば引数の中にデータ型を書いて出力する。

function func1(value: str) {
    return `Output: ${value}`
}

これに対して、Pythonの場合はpydantic.BaseModelを継承したモデルをクラスとして定義し、プロパティにデータ型を書いて型定義をする流れになる。

(2) `main.py`にルーティングを設定する

main.py

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from databases import Database
from .model import Task

app = FastAPI()
database = Database('sqlite:///tasks.db')

# データベースに接続する
@app.on_event("startup")
async def startup():
    await database.connect()

# データベースとの接続を遮断する
@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

@app.post("/tasks/", response_model=Task)
async def create_task(task: Task):
    query = "INSERT INTO tasks (title, description, completed) VALUES (:title, :description, :completed)"
    values = task.dict()
    task_id = await database.execute(query=query, values=values)
    response = {"id": task_id, **values}
    return JSONResponse(content=response, status_code=201)

@app.get("/tasks/", response_model=List[Task])
async def read_tasks(skip: int = 0, limit: int = 100):
    query = "SELECT * FROM tasks ORDER BY id DESC LIMIT :limit OFFSET :skip"
    tasks = await database.fetch_all(query=query, values={"skip": skip, "limit": limit})
    return tasks

@app.get("/tasks/{task_id}", response_model=Task)
async def read_task(task_id: int):
    query = "SELECT * FROM tasks WHERE id = :id"
    task = await database.fetch_one(query=query, values={"id": task_id})
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

@app.put("/tasks/{task_id}", response_model=Task)
async def update_task(task_id: int, task: Task):
    query = """
        UPDATE tasks SET title = :title, description = :description, completed = :completed
        WHERE id = :id
    """
    values = {"id": task_id, **task.dict()}
    result = await database.execute(query=query, values=values)
    if not result:
        raise HTTPException(status_code=404, detail="Task not found")
    return {**values}

@app.delete("/tasks/{task_id}")
async def delete_task(task_id: int):
    query = "DELETE FROM tasks WHERE id = :id"
    result = await database.execute(query=query, values={"id": task_id})
    if not result:
        raise HTTPException(status_code=404, detail="Task not found")
    return {"message": "Task deleted successfully"}

main.pyでは、FastAPIアプリケーションを作成し、SQLiteデータベースに接続している。また、Taskモデルを定義し、HTTPリクエストとレスポンスにこれを応用する。

ルーティングの部分では、HTTPメソッドとエンドポイントに応じて処理を実装する。具体的には、以下のエンドポイントが含まれている。

POSTメソッド：タスクを新規作成する。

@app.post("/tasks/", response_model=Task)
async def create_task(task: Task):
    query = "INSERT INTO tasks (title, description, completed) VALUES (:title, :description, :completed)"
    values = task.dict()
    task_id = await database.execute(query=query, values=values)
    response = {"id": task_id, **values}
    return JSONResponse(content=response, status_code=201)

GETメソッド：すべてのタスクを取得する。

@app.get("/tasks/{task_id}", response_model=Task)
async def read_task(task_id: int):
    query = "SELECT * FROM tasks WHERE id = :id"
    task = await database.fetch_one(query=query, values={"id": task_id})
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

GETメソッド(2)：/tasks/{task_id}にて、指定されたIDのタスクを取得する。

@app.get("/tasks/{task_id}", response_model=Task)
async def read_task(task_id: int):
    query = "SELECT * FROM tasks WHERE id = :id"
    task = await database.fetch_one(query=query, values={"id": task_id})
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

PUTメソッド：/tasks/{task_id}で指定されたIDのタスクを更新する。

@app.put("/tasks/{task_id}", response_model=Task)
async def update_task(task_id: int, task: Task):
    query = """
        UPDATE tasks SET title = :title, description = :description, completed = :completed
        WHERE id = :id
    """
    values = {"id": task_id, **task.dict()}
    result = await database.execute(query=query, values=values)
    if not result:
        raise HTTPException(status_code=404, detail="Task not found")
    return {**values}

DELETEメソッド：/tasks/{task_id}で、指定されたIDのタスクを削除する。

@app.delete("/tasks/{task_id}")
async def delete_task(task_id: int):
    query = "DELETE FROM tasks WHERE id = :id"
    result = await database.execute(query=query, values={"id": task_id})
    if not result:
        raise HTTPException(status_code=404, detail="Task not found")
    return {"message": "Task deleted successfully"}

(3) `init.py`の設定

現時点で、ディレクトリには以下のファイルが含まれている。

- main.py
- model.py

Pythonファイルをパッケージやライブラリとして活用する場合、同じディレクトリに__init__.pyを設置する。このとき、__init__.pyは空にしておく。Pythonファイルがパッケージやライブラリとして認識されるためには、空の__init__.pyを設置しなければならないのだ。

再度main.pyのソースコードを確認すると、以下のようなimport文が書かれていることがわかる。

from .model import Task

こちらのimport文は、現在のディレクトリ内のmodel.pyファイルからTaskクラスをインポートしている。model.pyをパッケージとして活用したい場合に、空の__init__.pyを同じディレクトリに新規作成する。

ここで、.は相対インポートを表しており、現在のディレクトリを起点として同じディレクトリにあるmodel.pyファイルを指定しています。

Taskクラスは、おそらくアプリケーションの中で使用されるクラスの1つであり、model.pyファイルに定義されているのが想定される。相対インポートを使用することで、ファイルの構成が変更されたときに、Taskクラスをインポートする際に発生する問題を回避できる。

なお、このPythonのプログラムは、from文によってTaskクラスのみをインポートしているため、model.pyファイルで定義されている他のクラスや関数はインポートされない。

(4) アプリケーションの起動

uvicornコマンドを使用してアプリケーションを起動する。

uvicorn main:app --reload

これで、FastAPIアプリケーションが起動する。また、SQLiteデータベースの作成は以下のコマンドで実行できる。

sqlite3 tasks.db

上述のコマンドを実行すると、SQLiteのコマンドラインインターフェイスが開く。以下のSQLコマンドを実行して、tasksテーブルを作成する。

CREATE TABLE tasks (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    description TEXT,
    completed BOOLEAN NOT NULL
);

(4)の別解

FastAPIのアプリケーションでSQLiteデータベースを作成する場合、SQLコマンドをファイルに書いて実行できる。

最初に、SQLコマンドを書いたファイルを作成する。例えば、create_table.sqlを作成して以下のようにSQLコマンドを書く。このとき、ファイルのディレクトリはmain.pyと同じ場所に置く。

create_table.sql

CREATE TABLE tasks (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    description TEXT,
    completed BOOLEAN NOT NULL
);

main.pyに、ファイルからSQLコマンドを読み込む処理を追加する。以下のように、with open()でファイルを開き、read()で内容を読み込む。

main.py

import sqlite3

db = sqlite3.connect('tasks.db')
with open('create_table.sql') as f:
    db.execute(f.read())

execute()メソッドでSQLコマンドが実行される。ファイルに複数のSQLコマンドが含まれている場合は、;で区切って1つの文字列にまとめる。

main.py

import sqlite3

db = sqlite3.connect('tasks.db')
with open('create_table.sql') as f:
    sql_commands = f.read()
db.executescript(sql_commands)

以上が、FastAPIとSQLiteを使用してタスクアプリのREST APIを作成する手順になる。

参考記事

GitHubで編集を提案

Discussion

hotta

役に立つ記事をありがとうございます。
『uvicorn（アッバーコーン）は』とありますが、Everything You Need to Know About Uvicornによると Uvicorn, pronounced “you-vee-corn”となっているようです。読み方は重要だと思い、コメントさせていただきました。

はじめに

本記事の対象読者

用語解説

FastAPI

uvicorn

SQLite

pydantic

実際の手順

(1) FastAPIのインストール

(2) APIのエンドポイントを定義する

(3) APIのエンドポイントにアクセスする

(4) main.pyでFastAPIのアプリサーバを構築する

タスクアプリのREST APIを開発する

手順

(1) model.pyの新規作成

(2) main.pyにルーティングを設定する

(3) __init__.pyの設定

(4) アプリケーションの起動

(4)の別解

参考記事

Discussion

(4) `main.py`でFastAPIのアプリサーバを構築する

(1) `model.py`の新規作成

(2) `main.py`にルーティングを設定する

(3) `init.py`の設定