Getting Started With MongoDB and FastAPI

1
git clone git@github.com:mongodb-developer/mongodb-with-fastapi.git

1
cd mongodb-with-fastapi
2
pip install -r requirements.txt

1
export MONGODB_URL="mongodb+srv://<username>:<password>@<url>/<db>?retryWrites=true&w=majority"

1
uvicorn app:app --reload

1
client = motor.motor_asyncio.AsyncIOMotorClient(os.environ["MONGODB_URL"])
2
db = client.get_database("college")
3
student_collection = db.get_collection("students")

1
# Represents an ObjectId field in the database.
2
# It will be represented as a `str` on the model so that it can be serialized to JSON.
3
PyObjectId = Annotated[str, BeforeValidator(str)]

1
class StudentModel(BaseModel):
2
    """
3
    Container for a single student record.
4
    """
5
6
    # The primary key for the StudentModel, stored as a `str` on the instance.
7
    # This will be aliased to `_id` when sent to MongoDB,
8
    # but provided as `id` in the API requests and responses.
9
    id: Optional[PyObjectId] = Field(alias="_id", default=None)
10
    name: str = Field(...)
11
    email: EmailStr = Field(...)
12
    course: str = Field(...)
13
    gpa: float = Field(..., le=4.0)
14
    model_config = ConfigDict(
15
        populate_by_name=True,
16
        arbitrary_types_allowed=True,
17
        json_schema_extra={
18
            "example": {
19
                "name": "Jane Doe",
20
                "email": "jdoe@example.com",
21
                "course": "Experiments, Science, and Fashion in Nanophotonics",
22
                "gpa": 3.0,
23
            }
24
        },
25
    )

1
class UpdateStudentModel(BaseModel):
2
    """
3
    A set of optional updates to be made to a document in the database.
4
    """
5
6
    name: Optional[str] = None
7
    email: Optional[EmailStr] = None
8
    course: Optional[str] = None
9
    gpa: Optional[float] = None
10
    model_config = ConfigDict(
11
        arbitrary_types_allowed=True,
12
        json_encoders={ObjectId: str},
13
        json_schema_extra={
14
            "example": {
15
                "name": "Jane Doe",
16
                "email": "jdoe@example.com",
17
                "course": "Experiments, Science, and Fashion in Nanophotonics",
18
                "gpa": 3.0,
19
            }
20
        },
21
    )

1
class StudentCollection(BaseModel):
2
    """
3
    A container holding a list of `StudentModel` instances.
4
5
    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/)
6
    """
7
8
    students: List[StudentModel]

1
@app.post(
2
    "/students/",
3
    response_description="Add new student",
4
    response_model=StudentModel,
5
    status_code=status.HTTP_201_CREATED,
6
    response_model_by_alias=False,
7
)
8
async def create_student(student: StudentModel = Body(...)):
9
    """
10
    Insert a new student record.
11
12
    A unique `id` will be created and provided in the response.
13
    """
14
    new_student = await student_collection.insert_one(
15
        student.model_dump(by_alias=True, exclude=["id"])
16
    )
17
    created_student = await student_collection.find_one(
18
        {"_id": new_student.inserted_id}
19
    )
20
    return created_student

1
@app.get(
2
    "/students/",
3
    response_description="List all students",
4
    response_model=StudentCollection,
5
    response_model_by_alias=False,
6
)
7
async def list_students():
8
    """
9
    List all of the student data in the database.
10
11
    The response is unpaginated and limited to 1000 results.
12
    """
13
    return StudentCollection(students=await student_collection.find().to_list(1000))

1
@app.get(
2
    "/students/{id}",
3
    response_description="Get a single student",
4
    response_model=StudentModel,
5
    response_model_by_alias=False,
6
)
7
async def show_student(id: str):
8
    """
9
    Get the record for a specific student, looked up by `id`.
10
    """
11
    if (
12
        student := await student_collection.find_one({"_id": ObjectId(id)})
13
    ) is not None:
14
        return student
15
16
    raise HTTPException(status_code=404, detail=f"Student {id} not found")

1
@app.put(
2
    "/students/{id}",
3
    response_description="Update a student",
4
    response_model=StudentModel,
5
    response_model_by_alias=False,
6
)
7
async def update_student(id: str, student: UpdateStudentModel = Body(...)):
8
    """
9
    Update individual fields of an existing student record.
10
11
    Only the provided fields will be updated.
12
    Any missing or `null` fields will be ignored.
13
    """
14
    student = {
15
        k: v for k, v in student.model_dump(by_alias=True).items() if v is not None
16
    }
17
18
    if len(student) >= 1:
19
        update_result = await student_collection.find_one_and_update(
20
            {"_id": ObjectId(id)},
21
            {"$set": student},
22
            return_document=ReturnDocument.AFTER,
23
        )
24
        if update_result is not None:
25
            return update_result
26
        else:
27
            raise HTTPException(status_code=404, detail=f"Student {id} not found")
28
29
    # The update is empty, but we should still return the matching document:
30
    if (existing_student := await student_collection.find_one({"_id": id})) is not None:
31
        return existing_student
32
33
    raise HTTPException(status_code=404, detail=f"Student {id} not found")

1
@app.delete("/students/{id}", response_description="Delete a student")
2
async def delete_student(id: str):
3
    """
4
    Remove a single student record from the database.
5
    """
6
    delete_result = await student_collection.delete_one({"_id": ObjectId(id)})
7
8
    if delete_result.deleted_count == 1:
9
        return Response(status_code=status.HTTP_204_NO_CONTENT)
10
11
    raise HTTPException(status_code=404, detail=f"Student {id} not found")