教程：使用 FastAPI 构建应用程序

# 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] = None    email: Optional[EmailStr] = None    course: Optional[str] = None    gpa: Optional[float] = None    model_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    """
    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 = student.model_dump(by_alias=True, exclude=["id"])    result = await student_collection.insert_one(new_student)    new_student["_id"] = result.inserted_id
    return new_student

@app.get(    "/students/",    response_description="List all students",    response_model=StudentCollection,    response_model_by_alias=False,)async def list_students():    """    List all 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 student
    raise 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_result        else:            raise HTTPException(status_code=404, detail=f"Student {id} not found")
    # The update is empty, so return the matching document:    if (existing_student := await student_collection.find_one({"_id": ObjectId(id)})) is not None:        return existing_student
    raise 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")