前回は Docker による FARM(FastAPI, React, MongoDB)の構築を行いました. これをベースにさまざまな Web アプリケーションの作成が可能です. 今回は基礎的な CRUD を行う ToDo アプリケーションの作成を目標にバックエンドの実装を行っていきます.

データベースへの接続

Python で MongoDB を扱うためのドライバには pymongo(公式推奨)と motorがあります. motor は MongoDB の非同期ドライバであり, 内部実装に pymongo が用いられています. どちらを使用しても問題ありませんが, 今回は単純な CRUD を行うために pymongo を使用します.

まずは./api/database/todo.pyを作成し DB クライアントとコレクションを初期化する処理を記述していきます.

# ./api/database/todo.py

from pymongo import MongoClient

HOST = 'mongo_db'
PORT = 27017
USERNAME = 'root'
PASSWORD = 'password'
DATABASE = 'todo_db'

client = MongoClient(HOST, PORT, username=USERNAME, password=PASSWORD)
db = client[DATABASE]


def make_todos():
    for i in range(10):
        db.todos.insert_one({
            'title': f'Todo {i}',
            'description': f'Todo {i} description',
            'completed': False
        })


if __name__ == '__main__':
    make_todos()

api コンテナの shell に入り, python database/todo.pyを実行すると, DB にデータが挿入されます.

python database/todo.py

Mongo Express(http://localhost:8081/db/todo_db/todos) にアクセスしてみるとドキュメントが 10 個作成されていることが確認できます.

MongoDB との接続は確認できましたので, 作成されたドキュメントは削除してしまっても問題ありません.

Pydantic モデルの作成

MongoDB はスキーマレスな DB であり自由度が高い反面, データの構造が複雑になるとバリデーションが難しくなります. そこで Pydantic を用いてスキーマを作成し, バリデーションを行います.

./api/schema/todo.pyを作成し, ToDo ドキュメントのスキーマを定義します.

# ./api/schema/todo.py

from typing import Union
from pydantic import BaseModel


class BaseTodo(BaseModel):
    title: str
    description: Union[str, None] = None
    completed: bool = False


class Todo(BaseTodo):
    _id: str


class CreateTodo(BaseTodo):
    pass

CRUD とルーティング

Create の実装とルーティング

MongoDB で CRUD を行う関数を作成してきます. まずは./api/crud/todo.pyを作成し, Create の実装を行います.

# ./api/crud/todo.py

from database.todo import db
from schema.todo import Todo, CreateTodo

def create_todo(todo: CreateTodo) -> Todo:
    db.todos.insert_one(todo.dict())
    return todo

続いてルーティングを行っていきます. ./api/routers/todo.pyを作成し, ルーティングを記述します.

# ./api/router/todo.py

import crud.todo as todo_crud
from fastapi import APIRouter
from schema.todo import CreateTodo, Todo

router = APIRouter()

@router.post('/todos')
def create_todo(todo: CreateTodo):
    return todo_crud.create_todo(todo)

最後に./api/main.pyにルーティングを追加します.

# ./api/main.py

import uvicorn
from fastapi import FastAPI
from router.todo import router as todo_router


app = FastAPI()

app.include_router(todo_router)

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


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

Create の実装とルーティングが完了しました. api サーバを起動していない場合は api コンテナの shell に入り, uvicorn main:app --reloadを実行してサーバーを起動します.

早速 Swagger UI(http://localhost:8000/docs)にアクセスしてみましょう.

ToDo の POST と POST 用のスキーマが作成されていることが確認できます. 実際に POST リクエストを送ってみましょう.

curl -X 'POST' \
  'http://localhost:8000/todos' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "title": "ゴミを捨てる",
  "description": "燃えるゴミ",
  "completed": false
}'

しっかりと ToDo ドキュメントが作成されていることが確認できました.

Read, Update, Delete の実装とルーティング

Create と同様に, Read, Update, Delete の実装とルーティングを行います. ./api/crud/todo.pyに実装を追加します.

# ./api/crud/todo.py

from bson import ObjectId
from database.todo import db
from schema.todo import Todo, CreateTodo, CreateTodoResponse


def create_todo(todo: CreateTodo) -> CreateTodoResponse:
    id = db.todos.insert_one(todo.dict())
    return CreateTodoResponse(id=str(id.inserted_id), **todo.dict())


def get_todos() -> list[Todo]:
    todos = list(db.todos.find())
    for idx, todo in enumerate(todos):
        todo['id'] = str(todo['_id'])
        todos[idx] = todo
    return todos


def update_todo(todo_id: str, todo: CreateTodo) -> CreateTodoResponse:
    db.todos.update_one({'_id': ObjectId(todo_id)}, {'$set': todo.dict()})
    return CreateTodoResponse(id=todo_id, **todo.dict())


def delete_todo(todo_id: str):
    db.todos.delete_one({'_id': ObjectId(todo_id)})
    return {'id': todo_id, 'message': 'Todo deleted successfully'}

続いて./api/router/todo.pyにルーティングを追加します.

# ./router/todo.py

import crud.todo as todo_crud
from fastapi import APIRouter
from schema.todo import CreateTodo, CreateTodoResponse, Todo

router = APIRouter()


@router.post('/todos', response_model=CreateTodoResponse)
def create_todo(todo: CreateTodo):
    return todo_crud.create_todo(todo)


@router.get('/todos', response_model=list[Todo])
def read_todos():
    return todo_crud.get_todos()


@router.put('/todos/{todo_id}')
def update_todo(todo_id: str, todo: CreateTodo):
    return todo_crud.update_todo(todo_id, todo)


@router.delete('/todos/{todo_id}')
def delete_todo(todo_id: str):
    return todo_crud.delete_todo(todo_id)

スキーマは以下のように定義しました.

# ./api/schema/todo.py

from typing import Union

from pydantic import BaseModel


class BaseTodo(BaseModel):
    title: str
    description: Union[str, None] = None
    completed: bool = False


class Todo(BaseTodo):
    id: str


class CreateTodo(BaseTodo):
    pass


class CreateTodoResponse(CreateTodo):
    id: str


class DeleteTodoResponse():
    id: str
    message: str

これで一通りの CRUD 操作ができるようになりました. api サーバを再起動して, Swagger UI(http://localhost:8000/docs)にアクセスしてみましょう.

まとめ

今回は FastAPI による API サーバの実装を行いました. Django のようなフルスタックなフレームワークではありませんが, 非常にシンプルで導入と実装が容易です. もし, Django のようなフルスタックなフレームワークを使いたい場合は, Django REST Framework を使うと良いでしょう. しかし今回のようなシンプルな API サーバを実装する場合は, FastAPI は非常に良い選択肢だと思います.