40岁的中年男子使用FastAPI和MongoDB连结的笔记①

关于本文

這篇文章是我作為一個剛開始學習編程的新手，在學習過程中對於一些生詞和概念進行了簡明易懂的摘要。如果有任何不正之處，請在評論區告訴我，我將不勝感激。

创建环境文件

在项目的根目录下创建一个.env文件。
然后从MongoDB复制Connect路径。
将部分更改为自己创建的内容。

MONGO_API_KEY=mongodb+srv://XXXXXXXXXXXXX

下一步，在项目的根目录下创建database.py。
再创建一个schemas.py。

数据库.py

在这里编写与MongoDB集成的处理代码。

from decouple import config
from typing import Union # 選択肢が２つ以上の場合に使える
import motor.motor_asyncio # こちらでMongoDB連携のために使う

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

insert_oneの返り値 insertResult(インスタンス)返り値になる。insert_idと言う属性が付与され、追加したIDを取得される。
insert_oneで返されたinsertResult(インスタンス)はオブジェクトクラスから生成されたインスタンスなので単純なstring型にならない(タイムスタンプなど色々入っている)

find_oneでidを取得 _idで格納されている。

find_oneのリターンのタイプは

ある場合はドキュメントを返して
無い場合はnoneを返す

schema.py

定义传递到终端点的数据类型，以定义响应的数据类型。

from pydantic import BaseModel # 型を定義するために使う

# エンドポイント返すデータ型
class Todo(BaseModel):
    id: str
    title: str
    description: str

# 新規作成に渡すデータ型
class TodoBody(BaseModel):
    title: str
    description: str

# mainのrootのデータ型
class SuccessMsg(BaseModel):
    message: str

route_todo.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 # database.pyで作った関数
from starlette.status import HTTP_201_CREATED # ステータスコードを上書きするのに必要

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
    # db_create_todoでFalseの場合の例外処理
    raise HTTPException(
        status_code=404, detail="Create task failed"
    )

主要的.py文件

from fastapi import FastAPI
from routers import route_todo # 作ったroutersのroute_todo
from schemas import SuccessMsg

app = FastAPI()
app.include_router(route_todo.router) # appにルートを追加

@app.get("/", response_model=SuccessMsg)
def root():
    return {"message": "利樹大好きだよ～～ん"}

以下是一个汉语本地化的选项：

参考如下: