はじめに
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の生成や取得などがおこなえるので、動作の確認を行ってください。
