はじめに

FastAPIは、Pythonの型ヒントに基づいてAPIを構築するためのWebフレームワークです。DBにRDSを用いる場合の情報はたくさんあるのですが、DBに MongoDB を用いた場合の情報が少なかったので、記述します。

MongoDB へのアクセスライブラリとして beanie を用いました。
Beanie は、Motor とPydantic に基づく、MongoDB用の非同期Pythonオブジェクトドキュメントマッパー（ODM）です。
ODMですので、FastAPIのバリデーションや型ヒントとの相性がいいです。

前提条件

Python 3.9以上
MongoDB 4.x
- この記事では DB 名を database_name としています

セットアップ

私は、pyenv と pipenv を利用してPythonの仮想環境を構築しています。使い慣れている仮想環境をお持ちの場合、適宜読み替えてください。

$ pyenv local 3.9.10
$ python --version
Python 3.9.10
    
$ pipenv install fastapi                                           
$ pipenv install uvicorn
$ pipenv install python-dotenv

// beanieは、mongoのODMライブラリ
$ pipenv install beanie
$ pipenv install 'pydantic[email]'

ファイルの構成

.
├── .env
├── database.py
├── main.py
└── schemas.py

.env ファイル

mongo サーバにアクセスするURLは、.env ファイルに格納しています。値の取得は python-dotenv ライブラリを利用しています。

MONGO_URL=mongodb://foo:foo_password@localhost:27001/

main.py

main.py は、unicorn から呼び出す FastAPI のメインです。

from datetime import datetime

from fastapi import Depends, FastAPI, HTTPException
from beanie import PydanticObjectId, WriteRules
    
import schemas
from database import init_db
    
app = FastAPI()

# FastAPI起動時にDBにアクセスする環境を初期化しています    
@app.on_event("startup")
async def start_db():
    await init_db()

# ユーザを作成します。     
# response_model で WebAPIのレスポンスの型を指定しています。
@app.post("/users", response_model=schemas.User)
async def create_user(user: schemas.UserCreate):
    
    user = await user.save(link_rule=WriteRules.WRITE) # Itemsも書き込む
    
    if len(user.items) > 0:
        for item in user.items:
            item.owner_id = user.id
    resp_user = await user.save(link_rule=WriteRules.WRITE)
    return resp_user

# idを指定してユーザを取得します    
@app.get("/users/{id}", response_model=schemas.User)
async def get_user(id: PydanticObjectId):
    user = await schemas.User.get(id)
    return user

# ユーザを更新します
@app.put("/users", response_model=schemas.User)
# 更新専用の UserUpdate クラスを用いています。
async def update_user(user: schemas.UserUpdate):
    current_user = await schemas.UserUpdate.get(user.id)
    if current_user is None:
        raise HTTPException(status_code=404, detail="User not found")

    # 更新処理では必要な属性だけを置き換えています
    current_user.updated_at = datetime.now()
    if user.password:
        current_user.password = user.password
    if user.email:
        current_user.email = user.email
    ret_user = await current_user.save()
    return ret_user

# ユーザのリストを取得します
@app.get("/users/", response_model=list[schemas.User])
async def read_users(skip: int = 0, limit: int = 100):
    users = await schemas.User.find().skip(skip).limit(limit).to_list()
    return users

# Itemを生成します。
@app.post("/items", response_model=schemas.Item)
async def create_item(item: schemas.ItemCreate):
    user = await schemas.User.get(item.owner_id)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")

    user.items.append(item)
    await user.save(link_rule=WriteRules.WRITE)
    return item

database.py

database.pyは、beanie や motor に関する初期化をおこなっています。
init_beanieの引数の database で MongoDB の DB 名を指定してください。

import os
    
from beanie import init_beanie
import motor.motor_asyncio
    
import schemas

async def init_db():
    # MONGO_URLは .env ファイルに定義しています 
    client = motor.motor_asyncio.AsyncIOMotorClient(os.getenv('MONGO_URL'))
    # MongoDBのDBを指定してください。
    await init_beanie(database=client.database_name,
       # 利用したい ODM を指定します
       document_models=[ 
           schemas.UserBase,schemas.User,schemas.UserCreate,schemas.UserUpdate,
           schemas.ItemBase,schemas.ItemCreate,schemas.Item,schemas.ItemUpdate
       ])

schema.py

schema.py は、Pydantic形式で記述する beanie の ODMモデルを定義しています。

from datetime import datetime
from typing import Optional

from beanie import Document, Link, PydanticObjectId
from pydantic import EmailStr, Field

# Itemの共通属性
class ItemBase(Document):
    # Pydantic では、':' はタイプを宣言するとき
    title: str
    # Pydantic では、'=' は値を設定するとき
    description: Optional[str] = None
    owner_id: Optional[PydanticObjectId]

    created_at: datetime = Field(default_factory=datetime.now)
    updated_at: datetime = Field(default_factory=datetime.now)

    class Settings:     # Settingsは beanie 用
        name = "items"  # mongoのコレクション名を指定

    class Config:       # Configは Pydantic 用
        # orm_mode=Trueとすることで、Item.id というアクセス可能。
        # ゆえに FastAPIの response_model に指定できます
        orm_mode = True

# 生成時と通常アクセス時とのクラスは分けておくのがベター
class ItemCreate(ItemBase):
    pass
class Item(ItemBase):
    created_at: Optional[datetime]
    updated_at: Optional[datetime]
    class Config:
        schema_extra = {
            "example": {
                "title": "book",
                "description": "ストレングス・ファインダー",
            }
        }

class ItemUpdate(Item):
    id: PydanticObjectId    # 更新は、idが必須


# User の共通属性
class UserBase(Document):
    email: EmailStr = Field(...)
    items: Optional[list[Link[Item]]] = []      # Itemのリレーション
    created_at: datetime = Field(default_factory=datetime.now)
    updated_at: datetime = Field(default_factory=datetime.now)

    class Settings:
        name = "users"

    class Config:       # Config は Pydantic 用
        orm_mode = True

# 生成時にはパスワードが必須。わかりやすい
class UserCreate(UserBase):
    # セキュリティを確保するため、パスワード属性は UserCreate クラスで定義し
    # Userクラスには定義しない。
    password: str = Field(...)


# Userクラスに passwordを含めない。そうすることでデータ読み取り時に
# パスワードが送信されることを防ぐ。
class User(UserBase):
    is_active: bool = False
    created_at: Optional[datetime]
    updated_at: Optional[datetime]

    class Config:       # Config は Pydantic 用
        schema_extra = {
            "example": {
                "email": "taro.example.com",
                "is_active": True,
                "items": [ { "title": "タイトル" } ]
            }
        }

class UserUpdate(User):
    id: PydanticObjectId    # 更新は、idが必須
    password: Optional[str] # パスワードはオプション

起動方法

$ pipenv shell
$ uvicorn main:app --port 3000 --reload
.....

動作の確認

FastAPIはOpenAPIのインタフェースを持っているので、ブラウザを用いて各APIの動作確認を行うことができます。

ブラウザで、http://localhost:3000/docs にアクセスすると、下記の画面が表示されます。

このページから、Userの生成や取得などがおこなえるので、動作の確認を行ってください。

参考資料