样例如下

import os
from typing import Optional, Listfrom fastapi import FastAPI, Body, HTTPException, status
from fastapi.responses import Response
from pydantic import ConfigDict, BaseModel, Field, EmailStr
from pydantic.functional_validators import BeforeValidatorfrom typing_extensions import Annotatedfrom bson import ObjectId
import motor.motor_asyncio
from pymongo import ReturnDocumentapp = FastAPI(title="Student Course API",summary="A sample application showing how to use FastAPI to add a ReST API to a MongoDB collection.",
)
client = motor.motor_asyncio.AsyncIOMotorClient(os.environ["MONGODB_URL"])
db = client.college
student_collection = db.get_collection("students")# Represents an ObjectId field in the database.
# It will be represented as a `str` on the model so that it can be serialized to JSON.
PyObjectId = Annotated[str, BeforeValidator(str)]class StudentModel(BaseModel):"""Container for a single student record."""# The primary key for the StudentModel, stored as a `str` on the instance.# This will be aliased to `_id` when sent to MongoDB,# but provided as `id` in the API requests and responses.id: Optional[PyObjectId] = Field(alias="_id", default=None)name: str = Field(...)email: EmailStr = Field(...)course: str = Field(...)gpa: float = Field(..., le=4.0)model_config = ConfigDict(populate_by_name=True,arbitrary_types_allowed=True,json_schema_extra={"example": {"name": "Jane Doe","email": "jdoe@example.com","course": "Experiments, Science, and Fashion in Nanophotonics","gpa": 3.0,}},)class UpdateStudentModel(BaseModel):"""A set of optional updates to be made to a document in the database."""name: Optional[str] = Noneemail: Optional[EmailStr] = Nonecourse: Optional[str] = Nonegpa: Optional[float] = Nonemodel_config = ConfigDict(arbitrary_types_allowed=True,json_encoders={ObjectId: str},json_schema_extra={"example": {"name": "Jane Doe","email": "jdoe@example.com","course": "Experiments, Science, and Fashion in Nanophotonics","gpa": 3.0,}},)class StudentCollection(BaseModel):"""A container holding a list of `StudentModel` instances.This exists because providing a top-level array in a JSON response can be a [vulnerability](https://haacked.com/archive/2009/06/25/json-hijacking.aspx/)"""students: List[StudentModel]@app.post("/students/",response_description="Add new student",response_model=StudentModel,status_code=status.HTTP_201_CREATED,response_model_by_alias=False,
)
async def create_student(student: StudentModel = Body(...)):"""Insert a new student record.A unique `id` will be created and provided in the response."""new_student = await student_collection.insert_one(student.model_dump(by_alias=True, exclude=["id"]))created_student = await student_collection.find_one({"_id": new_student.inserted_id})return created_student@app.get("/students/",response_description="List all students",response_model=StudentCollection,response_model_by_alias=False,
)
async def list_students():"""List all of the student data in the database.The response is unpaginated and limited to 1000 results."""return StudentCollection(students=await student_collection.find().to_list(1000))@app.get("/students/{id}",response_description="Get a single student",response_model=StudentModel,response_model_by_alias=False,
)
async def show_student(id: str):"""Get the record for a specific student, looked up by `id`."""if (student := await student_collection.find_one({"_id": ObjectId(id)})) is not None:return studentraise HTTPException(status_code=404, detail=f"Student {id} not found")@app.put("/students/{id}",response_description="Update a student",response_model=StudentModel,response_model_by_alias=False,
)
async def update_student(id: str, student: UpdateStudentModel = Body(...)):"""Update individual fields of an existing student record.Only the provided fields will be updated.Any missing or `null` fields will be ignored."""student = {k: v for k, v in student.model_dump(by_alias=True).items() if v is not None}if len(student) >= 1:update_result = await student_collection.find_one_and_update({"_id": ObjectId(id)},{"$set": student},return_document=ReturnDocument.AFTER,)if update_result is not None:return update_resultelse:raise HTTPException(status_code=404, detail=f"Student {id} not found")# The update is empty, but we should still return the matching document:if (existing_student := await student_collection.find_one({"_id": id})) is not None:return existing_studentraise HTTPException(status_code=404, detail=f"Student {id} not found")@app.delete("/students/{id}", response_description="Delete a student")
async def delete_student(id: str):"""Remove a single student record from the database."""delete_result = await student_collection.delete_one({"_id": ObjectId(id)})if delete_result.deleted_count == 1:return Response(status_code=status.HTTP_204_NO_CONTENT)raise HTTPException(status_code=404, detail=f"Student {id} not found")