PynamoDB + FastAPIを用いたAPI開発方法 メモ

• PynamoDBとFastAPIを用いて簡単なCRUD APIを作成したのでメモとして残しておく。

PynamoDB

• DynamoDB用Pythonインターフェイス
• DynamoDB APIを抽象化しており、比較的簡単にDynamoDB操作ができる。

事前準備

• こちらの手順で動作確認用のDynamoDBコンテナを起動する。

  • テーブル名などはtest_userとして修正する。

• 依存ライブラリインストール

  pynamodb
  fastapi
  uvicorn
  

コード

• app.py

from pynamodb.models import Model
from pynamodb.attributes import UnicodeAttribute,NumberAttribute
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
from pydantic import BaseModel
from typing import Optional
import uuid
import logging

# PynamoDB用データモデル
class UserModel(Model):
    # 接続先情報(クレデンシャルやテーブル情報)を定義
    class Meta:
        host = "http://localhost:8000"
        aws_access_key_id = 'dummy'
        aws_secret_access_key = 'dummy'
        region = 'ap-northeast-1'
        table_name = "test_user"
    # テーブル属性定義
    id = UnicodeAttribute(hash_key=True)
    email = UnicodeAttribute(null=True)
    
# FastAPIリクエストボディ用データモデル
class User(BaseModel):
    id: Optional[str] = None
    email: str
    
app = FastAPI()

# ユーザー登録API
@app.post("/users")
def regist_user(user: User):
    # ユーザーID生成
    user_id = str(uuid.uuid4()).replace("-","")
    regist_user = UserModel(user_id)
    regist_user.email = user.email
    # ユーザー登録
    regist_user.save()
    
    user.id = user_id
    return user

# ユーザー取得API
@app.get("/users/{user_id}")
def get_user(user_id: str):
    try:
        # ユーザー取得
        user = UserModel.get(user_id)
        logging.info(user)
        return user.attribute_values
    except UserModel.DoesNotExist:
        logging.error("User does not exist.")
        raise HTTPException(status_code=404, detail="user_not_found")


# ユーザー更新API
@app.put("/users/{user_id}")
def update_user(user_id:str, user: User):
    try:
        # 更新対象ユーザー取得
        target_user = UserModel.get(user_id)
        # ユーザー更新
        target_user.update(actions=[
            UserModel.email.set(user.email)
        ])
        return target_user.attribute_values
    except UserModel.DoesNotExist:
        logging.error("User does not exist.")
        raise HTTPException(status_code=404, detail="user_not_found")

# ユーザー削除API
@app.delete("/users/{user_id}")
def update_user(user_id:str):
    try:
        # 削除対象ユーザー取得
        target_user = UserModel.get(user_id)
        # ユーザー削除
        target_user.delete()
        return {}
    except UserModel.DoesNotExist:
        logging.error("User does not exist.")
        raise HTTPException(status_code=404, detail="user_not_found")    

動作確認

• API起動

  uvicorn app:app --reload --port 5000
  

• ユーザー登録API

  • リクエスト

    POST /users HTTP/1.1
    Host: 127.0.0.1:5000
    Content-Type: application/json
    Content-Length: 37
    
    {
        "email":"[e-mail]"
    }
    

  • レスポンス

    {
        "id": "87527962cbd14293943d323548f403da",
        "email": "[e-mail]"
    }
    

• ユーザー取得API

  • リクエスト

    GET /users/87527962cbd14293943d323548f403da HTTP/1.1
    Host: 127.0.0.1:5000
    

  • レスポンス

    {
        "email": "[e-mail]",
        "id": "87527962cbd14293943d323548f403da"
    }
    

• ユーザー更新API

  • リクエスト

    PUT /users/87527962cbd14293943d323548f403da HTTP/1.1
    Host: 127.0.0.1:5000
    Content-Type: application/json
    Content-Length: 44
    
    {
        "email":"[e-mail]"
    }
    

  • レスポンス

    {
        "email": "[e-mail]",
        "id": "87527962cbd14293943d323548f403da"
    }
    

• ユーザー削除API

  • リクエスト

    DELETE /users/87527962cbd14293943d323548f403da HTTP/1.1
    Host: 127.0.0.1:5000
    

  • レスポンス

    {}
    

参考情報

• PynamoDB