跳至主要内容

[note] beanie ODM

參考資料

搭配 FastAPI

程式碼

Python Beanie Playground @ GitLab

建立連線

# main.py

from contextlib import asynccontextmanager
from typing import Optional

from beanie import init_beanie, Document, Indexed
from fastapi import FastAPI
from motor.motor_asyncio import AsyncIOMotorClient

from python_beanie_sandbox.models.book import Book
from python_beanie_sandbox.models.product import Product

from python_beanie_sandbox.routes.book import router as book_router
from python_beanie_sandbox.routes.product import router as product_router


@asynccontextmanager
async def lifespan(app: FastAPI):
client = AsyncIOMotorClient("mongodb://localhost:27017/")
await init_beanie(database=client.get_database("playground"), document_models=[Book, Product])
try:
yield
finally:
client.close()

app = FastAPI(lifespan=lifespan)

app.include_router(book_router, prefix="/v1")
app.include_router(product_router, prefix="/v1")

定義 Schema (model)

  • id 可以用 PydanticObjectId
  • 使用 Settings.name 可以修改被建立在 MongoDB 中的 collection 名稱
  • 使用 Config.json_schema_extra 可以幫助產出的 Swagger 文件能提供 example request payload
# python_beanie_sandbox/models/book.py

from typing import Optional

from beanie import Document
from pydantic import BaseModel


from typing import Optional

from beanie import Document, PydanticObjectId

from pydantic import BaseModel


class Book(Document):
id: PydanticObjectId
title: str
author: str
published_year: int
reviews: list[str] = []

class Settings:
name = "books"

class Config:
json_schema_extra = {
"title": "Book",
"description": "A book with title, author, published year and reviews",
"example": {
"title": "The Great Gatsby",
"author": "F. Scott Fitzgerald",
"published_year": 1925,
"reviews": ["A classic!", "A must-read!"]
}
}

# 如果沒有要在 MongoDB 中建立對應的 collection,則一樣可以使用 BaseModel
class UpdateBook(BaseModel):
# 在新版的 fastAPI 中,如果欄位是 optional 的,需要給預設值 None
title: Optional[str] = None
author: Optional[str] = None
published_year: Optional[int] = None

class Config:
json_schema_extra = {
"title": "UpdateBook",
"description": "A book with title, author and published year",
"example": {
"title": "The Great Gatsby",
"author": "F. Scott Fitzgerald",
"published_year": 1925
}
}

提示

如果是 Model level 的資料驗證沒過,會直接噴 error 在 server 的 log 上。

定義路由

提示

如果是 Routes level 的資料驗證沒過,fastAPI 會回傳 422 Error: Unprocessable Entity,並帶上錯誤訊息。

from typing import List

from beanie import PydanticObjectId
from fastapi import APIRouter, HTTPException
from starlette import status

from python_beanie_sandbox.models.book import Book, UpdateBook, CreateBook
from python_beanie_sandbox.utils import pydantic_encoder

router = APIRouter(prefix="/books", tags=["book"])


@router.post("/", status_code=status.HTTP_201_CREATED, response_model=Book)
async def create_book(book_data: CreateBook):
book = Book(**book_data.model_dump(), id=PydanticObjectId())
await book.insert()
return book


@router.get("/", response_model=List[Book])
async def get_books():
return await Book.find_all().to_list()


@router.get("/{book_id}", response_model=Book)
async def get_book(book_id: PydanticObjectId):
book = await Book.get(book_id)
if book is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Book with id {book_id} not found")
return book


@router.patch("/{book_id}", response_model=Book)
async def patch_book(book_id: PydanticObjectId, update_book: UpdateBook):
book = await Book.get(book_id)
if book is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Book with id {book_id} not found")

book_data = pydantic_encoder.encode_input(update_book)
await book.update({"$set": book_data})
updated_book = await Book.get(book_id)
return updated_book


@router.delete("/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_book(book_id: PydanticObjectId):
book = await Book.get(book_id)
if book is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Book with id {book_id} not found")

await book.delete()
return None


@router.post("/{book_id}/reviews", response_model=Book)
async def add_book_review(book_id: PydanticObjectId, review: str):
book = await Book.get(book_id)
if book is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Book with id {book_id} not found")

await book.update({"$push": {"reviews": review}})
updated_book = await Book.get(book_id)
return updated_book