Getting Started with MongoDB and FastAPI

#Connecting to MongoDB

One of the very first things we do is connect to our MongoDB database.

1
client = motor.motor_asyncio.AsyncIOMotorClient(os.environ["MONGODB_URL"])
2
db = client.college

We're using the async motor driver to create our MongoDB client, and then we specify our database name college.

#The _id Attribute and ObjectIds

1
class PyObjectId(ObjectId):
2
    @classmethod
3
    def __get_validators__(cls):
4
        yield cls.validate
5
6
    @classmethod
7
    def validate(cls, v):
8
        if not ObjectId.is_valid(v):
9
            raise ValueError("Invalid objectid")
10
        return ObjectId(v)
11
12
    @classmethod
13
    def __modify_schema__(cls, field_schema):
14
        field_schema.update(type="string")

MongoDB stores data as BSON. FastAPI encodes and decodes data as JSON strings. BSON has support for additional non-JSON-native data types, including ObjectId which can't be directly encoded as JSON. Because of this, we convert ObjectIds to strings before storing them as the _id.

#Database Models

Many people think of MongoDB as being schema-less, which is wrong. MongoDB has a flexible schema. That is to say that collections do not enforce document structure by default, so you have the flexibility to make whatever data-modelling choices best match your application and its performance requirements. So, it's not unusual to create models when working with a MongoDB database. Our application has two models, the StudentModel and the UpdateStudentModel.

1
class StudentModel(BaseModel):
2
    id: PyObjectId = Field(default_factory=PyObjectId, alias="_id")
3
    name: str = Field(...)
4
    email: EmailStr = Field(...)
5
    course: str = Field(...)
6
    gpa: float = Field(..., le=4.0)
7
8
    class Config:
9
        allow_population_by_field_name = True
10
        arbitrary_types_allowed = True
11
        json_encoders = {ObjectId: str}
12
        schema_extra = {
13
            "example": {
14
                "name": "Jane Doe",
15
                "email": "jdoe@example.com",
16
                "course": "Experiments, Science, and Fashion in Nanophotonics",
17
                "gpa": "3.0",
18
            }
19
        }

This is the primary model we use as the response model for the majority of our endpoints.

I want to draw attention to the id field on this model. MongoDB uses _id, but in Python, underscores at the start of attributes have special meaning. If you have an attribute on your model that starts with an underscore, pydantic—the data validation framework used by FastAPI—will assume that it is a private variable, meaning you will not be able to assign it a value! To get around this, we name the field id but give it an alias of _id. You also need to set allow_population_by_field_name to True in the model's Config class.

We set this id value automatically to an ObjectId string, so you do not need to supply it when creating a new student.

1
class UpdateStudentModel(BaseModel):
2
    name: Optional[str]
3
    email: Optional[EmailStr]
4
    course: Optional[str]
5
    gpa: Optional[float]
6
7
    class Config:
8
        arbitrary_types_allowed = True
9
        json_encoders = {ObjectId: str}
10
        schema_extra = {
11
            "example": {
12
                "name": "Jane Doe",
13
                "email": "jdoe@example.com",
14
                "course": "Experiments, Science, and Fashion in Nanophotonics",
15
                "gpa": "3.0",
16
            }
17
        }

The UpdateStudentModel has two key differences from the StudentModel:

• It does not have an id attribute as this should never change.
• All fields are optional, so you only need to supply the fields you wish to update.

#Application Routes

Our application has five routes:

• POST / - creates a new student.
• GET / - view a list of all students.
• GET /{id} - view a single student.
• PUT /{id} - update a student.
• DELETE /{id} - delete a student.

1
@app.post("/", response_description="Add new student", response_model=StudentModel)
2
async def create_student(student: StudentModel = Body(...)):
3
    student = jsonable_encoder(student)
4
    new_student = await db["students"].insert_one(student)
5
    created_student = await db["students"].find_one({"_id": new_student.inserted_id})
6
    return JSONResponse(status_code=status.HTTP_201_CREATED, content=created_student)

1
@app.get(
    "/", response_description="List all students", response_model=List[StudentModel]
)
2
async def list_students():
3
    students = await db["students"].find().to_list(1000)
4
    return students

1
@app.get(
    "/{id}", response_description="Get a single student", response_model=StudentModel
)
2
async def show_student(id: str):
3
    if (student := await db["students"].find_one({"_id": id})) is not None:
4
        return student
5
6
    raise HTTPException(status_code=404, detail=f"Student {id} not found")

1
@app.put("/{id}", response_description="Update a student", response_model=StudentModel)
2
async def update_student(id: str, student: UpdateStudentModel = Body(...)):
3
    student = {k: v for k, v in student.dict().items() if v is not None}
4
5
    if len(student) >= 1:
6
        update_result = await db["students"].update_one({"_id": id}, {"$set": student})
7
8
        if update_result.modified_count == 1:
9
            if (
10
                updated_student := await db["students"].find_one({"_id": id})
11
            ) is not None:
12
                return updated_student
13
14
    if (existing_student := await db["students"].find_one({"_id": id})) is not None:
15
        return existing_student
16
17
    raise HTTPException(status_code=404, detail=f"Student {id} not found")

1
@app.delete("/{id}", response_description="Delete a student")
2
async def delete_student(id: str):
3
    delete_result = await db["students"].delete_one({"_id": id})
4
5
    if delete_result.deleted_count == 1:
6
        return JSONResponse(status_code=status.HTTP_204_NO_CONTENT)
7
8
    raise HTTPException(status_code=404, detail=f"Student {id} not found")