Getting Started With FastAPI and MongoDB

We all know that FastAPI is on the rise. According to survey data from Jetbrains, 20% of Python developers reported using FastAPI in 2022, up from 14% in 2021. This change was paired by slight declines in both Django (45% -> 39%) and Flask (46% -> 40%) over the same time period.

After building a full-stack app with Flask recently, I decided to give FastAPI a try by building a super simple API. After all, it is supposed to be fast for building APIs.

To get started, I followed along with this tutorial, but made some modifications to work with a data set that I thought was more fun and interesting - a sample data set you can access in MongoDB Atlas on shipwrecks.

To get started, I created a single file called main.py and imported the libraries I would be using:

from fastapi import FastAPI
from pydantic import BaseModel, Field
from bson import ObjectId
from typing import Optional, List
import motor.motor_asyncio

I imported FastAPI, some features from the Pydantic data validation framework, ObjectID to help with handling the IDs for my MongoDB documents, and Optional and List from the Python typing module, and Motor async library for MongoDB.

Next, I connected to MongoDB with Motor and declared my FastAPI app. Notice there's a placeholder where you'd need to add your own connection string if you give this a try with your own free MongoDB Atlas cluster.

client = motor.motor_asyncio.AsyncIOMotorClient("YOUR_CONNECTION_STRING")
db = client.sample_geospatial

app = FastAPI()

I then used this code (taken from the tutorial I mentioned above) to create the PyObjectId class, which will let us take the ObjectID BSON type in MongoDB and encode it as a string when used in this API.

class PyObjectId(ObjectId):
    @classmethod
    def __get_validators__(cls):
        yield cls.validate

    @classmethod
    def validate(cls, v):
        if not ObjectId.is_valid(v):
            raise ValueError("Invalid objectid")
        return ObjectId(v)

    @classmethod
    def __modify_schema__(cls, field_schema):
        field_schema.update(type="string")

Next up, models! Working from my shipwrecks sample data set, I created a class for shipwrecks, and provided configuration details for the class and a sample schema.

# Define model
# Set default value as None to avoid triggering error when there is no value for that field in MongoDB
class ShipwreckModel(BaseModel):
    id: Optional[PyObjectId] = Field(default_factory=PyObjectId, alias="_id")
    rcrd: Optional[str] = None
    vesslterms: Optional[str] = None
    feature_type: Optional[str] = None
    chart: Optional[str] = None
    latdec: Optional[float] = None
    londec: Optional[float] = None
    gp_quality: Optional[str] = None
    depth: Optional[float] = None
    sounding_type: Optional[str] = None
    quasou: Optional[str] = None
    watlev: Optional[str] = None

    class Config:
        allow_population_by_field_name = True
        arbitrary_types_allowed = True
        json_encoders = {ObjectId: str}
        schema_extra = {
            "_id": {
                "$oid": "578f6fa2df35c7fbdbaed8ec"
            },
            "recrd": "",
            "vesslterms": "",
            "feature_type": "Wrecks - Visible",
            "chart": "US,US,reprt,L-1453/14",
            "latdec": 18.5640431,
            "londec": -72.3525848,
            "gp_quality": "",
            "depth": 0,
            "sounding_type": "",
            "history": "",
            "quasou": "",
            "watlev": "always dry",
        }

I did run into some challenges when setting this up. At first, I did not mark the fields as optional and used Field like in the tutorial. So for example, a field would be: rcrd: str = Field(...). But then I started getting this error

So instead I reconfigured the class so that each field is optional and has a default value of None so an error isn't triggered when a field doesn't have a value in one of the Shipwreck documents. I imagine if I had a more straightforward dataset where every field had a corresponding value I would not need to use this workaround.

Now time to start pulling some data. I created a read route that lets you pull all of the shipwreck documents in the database (up to a list of 1000):

# Get all shipwrecks
@app.get(
    "/", response_description="List all shipwreckss", response_model=List[ShipwreckModel]
)
async def list_shipwrecks():
    shipwrecks = await db["shipwrecks"].find().to_list(1000)
    return shipwrecks

What I would like to do next is create additional read routes like the following:

# Get a random shipwreck
@app.get(
    "/random", response_description="Get a random shipwreck", response_model=List[ShipwreckModel]
)
async def random_shipwreck():
    random = await db["shipwrecks"].find_one({"depth": {'gte': 0.0}})
    return random

But this causes a different error I am still working through, likely related to the way I assigned a default None value in the field definitions for my ShipwreckModel class:

Next steps in my FastAPI explorations will be working through this error so I can create the some new read routes and also adding create, update, and delete routes to the API. Onward!

Check out the full code to go with this post here.

Cover image sources from Pexels.