8 Best Practices for Building FastAPI and MongoDB Applications

1
pip install motor[srv]

1
from fastapi import FastAPI
2
from motor.motor_asyncio import AsyncIOMotorClient
3
4
app = FastAPI()
5
6
# Load the MongoDB connection string from the environment variable MONGODB_URI
7
CONNECTION_STRING = os.environ['MONGODB_URI']
8
9
# Create a MongoDB client
10
client = AsyncIOMotorClient(CONNECTION_STRING)

1
from contextlib import asynccontextmanager
2
from logging import info @asynccontextmanager
3
async def db_lifespan(app: FastAPI):
4
    # Startup
5
    app.mongodb_client = AsyncIOMotorClient(CONNECTION_STRING)
6
    app.database = app.mongodb_client.get_default_database()
7
    ping_response = await app.database.command("ping")
8
    if int(ping_response["ok"]) != 1:
9
        raise Exception("Problem connecting to database cluster.")
10
    else:
11
        info("Connected to database cluster.")
12
    
13
    yield
14
15
    # Shutdown
16
    app.mongodb_client.close()
17
18
19
app: FastAPI = FastAPI(lifespan=db_lifespan)

1
class Profile(Document):
2
    """
3
    A profile for a single user as a Beanie Document.
4
5
    Contains some useful information about a person.
6
    """
7
8
    # Use a string for _id, instead of ObjectID:
9
    id: Optional[str] = Field(default=None, description="MongoDB document ObjectID")
10
    username: str
11
    birthdate: datetime
12
    website: List[str]
13
14
    class Settings:
15
        # The name of the collection to store these objects.
16
        name = "profiles"
17
18
# A sample path operation to get a Profile:
19
@app.get("/profiles/{profile_id}")
20
async def get_profile(profile_id: str) -> Profile:
21
    """
22
    Look up a single profile by ID.
23
    """
24
    # This API endpoint demonstrates using Motor directly to look up a single
25
    # profile by ID.
26
    profile = await Profile.get(profile_id)
27
    if profile is not None:
28
        return profile
29
    else:
30
        raise HTTPException(
31
            status_code=404, detail=f"No profile with id '{profile_id}'"
32
        )

1
@app.get("/people/{profile_id}")
2
async def read_item(profile_id: str) -> Profile:
3
    """ Use Beanie to look up a Profile. """
4
    profile = await Profile.get(profile_id)
5
    return profile

1
@app.get(
2
    "/people/{profile_id}",
3
    response_model=Profile,
4
)
5
async def read_item(profile_id: str) -> Mapping[str, Any]:
6
    # This API endpoint demonstrates using Motor directly to look up a single
7
    # profile by ID.
8
    #
9
    # It uses response_model (above) to tell FastAPI the schema of the data
10
    # being returned, but it returns a dict directly, so that conversion and
11
    # validation is done by FastAPI, meaning you don't have to copy values
12
    # manually into a Profile before returning it.
13
    profile = await app.profiles.find_one({"_id": profile_id})
14
    if profile is not None:
15
        return profile

1
# A Pydantic class modelling the *response* schema.
2
class Profile(BaseModel):
3
    """
4
    A profile for a single user.
5
    """
6
    id: Optional[str] = Field(
7
        default=None, description="MongoDB document ObjectID", alias="_id"
8
    )
9
    username: str
10
    residence: str
11
    current_location: List[float]
12
13
# A path operation that returns a Profile object as JSON:
14
@app.get(
15
    "/profiles/{profile_id}",
16
    response_model=Profile, # This tells FastAPI that the returned object must match the Profile schema.
17
)
18
async def get_profile(profile_id: str) -> Mapping[str, Any]:
19
    # Uses response_model (above) to tell FastAPI the schema of the data
20
    # being returned, but it returns a dict directly, so that conversion and
21
    # validation is done by FastAPI, meaning you don't have to copy values
22
    # manually into a Profile before returning it.
23
    profile = await app.profiles.find_one({"_id": profile_id})
24
    if profile is not None:
25
        return profile  # Return BSON document (Mapping). Conversion etc will be done automatically.
26
    else:
27
        raise HTTPException(
28
            status_code=404, detail=f"No profile with id '{profile_id}'"
29
        )