４０代おっさんFastAPI、MongoDB連携メモ②

关于本文

这篇文章是我作为一个编程初学者在学习过程中总结的一些易于理解的单词和概述。如果有任何不正确的地方，请在评论中告知我，谢谢。

获取任务列表

数据库.py

async def db_get_todos() -> list: # 一覧が返ってくるようにlist型を定義
    todos = [] # MongoDBから返ってくるデータを受ける配列
    # タスクの一覧を受け取る処理
    for todo in await collection_todo.find().to_list(length=100): # lengthで最大件数を定義
        todos.append(todo_serializer(todo)) # for文で展開した物をtodosに格納
    return todos

collection_todo.find()motorのメソッド(‘find()’)
‘find()’はmotorのインスタンスを作ってくれるもの

to_list実行するときに通信が行われるのでasyncが必要になる

路线_待办.py .py)

@router.get("/api/todo", response_model=List[Todo]) # 一覧取得なのでList型で返ってくる
async def get_todos():
    res = await db_get_todos()
    return res

获取特定信息

数据库.py

from bson import ObjectId

async def db_get_single_todo(id: str) -> Union[dict, bool]: # 取得したいidをstringで取得、返り値はdict,boolを返す
    todo = await collection_todo.find_one({"_id": ObjectId(id)}) # ObjectIdに変換
    if todo:
        return todo_serializer(todo)
    return False

from bson import ObjectIdMongo DBではBsonとうフォーマット保存されている

route_todo.py 的中文翻译为：路线待办事项.py

@router.get("/api/todo/{id}", response_model=Todo) # 一つしか返ってこない
async def get_single_todo(id: str):
    res = await db_get_single_todo(id)
    if res:
        return res
    raise HTTPException(
        status_code=404, detail=f"Task of ID:{id} doesn't exist"
    )

合并了上次的代码

数据库.py

from decouple import config
from typing import Union # 選択肢が２つ以上の場合に使える
import motor.motor_asyncio # こちらでMongoDB連携のために使う
from bson import ObjectId # MongoDB内でBsonフォーマット保存されているため

MONGO_API_KEY = config('MONGO_API_KEY') # configを使って環境変数読み込む

client = motor.motor_asyncio.AsyncIOMotorClient(MONGO_API_KEY) # MongoDB連携 APIのキーを渡す
database = client.API_DB # MongoDBの名前と一致
collection_todo = database.todo # MongoDBの名前と一致
collection_user = database.user # MongoDBの名前と一致

# insert_oneを辞書型で返すために必要な関数
def todo_serializer(todo) -> dict:
    return {
        "id": str(todo["_id"]),
        "title": todo["title"],
        "description": todo["description"]
    }

async def db_create_todo(data: dict) -> Union[dict, bool]: # Unionで選択肢をdict, bool選べるようにしている
    todo = await collection_todo.insert_one(data) # insert_oneの返り値 insertResult(インスタンス)返り値になる
    new_todo = await collection_todo.find_one({"_id": todo.inserted_id}) # find_oneでidを取得
    # find_oneのあるなし判定
    if new_todo:
        return todo_serializer(new_todo) 
    return False

async def db_get_todos() -> list: # 一覧が返ってくるようにlist型を定義
    todos = [] # MongoDBから返ってくるデータを受ける配列
    # タスクの一覧を受け取る処理
    for todo in await collection_todo.find().to_list(length=100): # lengthで最大件数を定義
        todos.append(todo_serializer(todo)) # for文で展開した物をtodosに格納
    return todos

async def db_get_single_todo(id: str) -> Union[dict, bool]: # 取得したいidをstringで取得、返り値はdict,boolを返す
    todo = await collection_todo.find_one({"_id": ObjectId(id)}) # ObjectIdに変換
    if todo:
        return todo_serializer(todo)
    return False

路线_待办.py

from fastapi import APIRouter # routerを作るのに必要
from fastapi import Response, Request, HTTPException # 11行目、21行目の関数
from fastapi.encoders import jsonable_encoder # jsonable_encoder型で受け取ったデータを辞書型にするために必要
from schemas import Todo, TodoBody # 型定義
from database import db_create_todo, db_get_todos, db_get_single_todo # database.pyで作った関数
from starlette.status import HTTP_201_CREATED # ステータスコードを上書きするのに必要
from typing import List # Listを使うため

router = APIRouter() # インスタンス化

@router.post("/api/todo", response_model=Todo)
async def create_todo(request: Request, response: Response, data:TodoBody): # 実行されたときの関数を定義
    todo = jsonable_encoder(data) # 受け取ったdataを辞書型にする
    res = await db_create_todo(todo)
    response.status_code = HTTP_201_CREATED # ステータスコードの上書き
    # jsonable_encoder型存在していたらデータを存在しなければnoneを返すので判定する
    if res:
        return res
    # noneの場合の例外処理
    raise HTTPException(
        status_code=404, detail="Create task failed"
    )

@router.get("/api/todo", response_model=List[Todo]) # 一覧取得なのでList型で返ってくる
async def get_todos():
    res = await db_get_todos()
    return res

@router.get("/api/todo/{id}", response_model=Todo) # 一つしか返ってこない
async def get_single_todo(id: str):
    res = await db_get_single_todo(id)
    if res:
        return res
    raise HTTPException(
        status_code=404, detail=f"Task of ID:{id} doesn't exist"
    )

请参考