[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